From 721ac7cc80c12d92de4f663f8a17d993eaf13fe9 Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Wed, 24 Jun 2026 15:53:22 +0200 Subject: [PATCH 1/9] refactor(router): extract encryptAndUpload from object.put Pull the envelope-crypto + upstream-PutObject sequence (DEK encrypt, KEK-version metadata tagging, client.PutObject, buffer release) out of object.put into a reusable encryptAndUpload(ctx, log) method. put keeps the detached-context setup, error-code mapping, response headers, and 200 write. No behaviour change. Prepares a single shared crypto path for buffered-multipart CompleteMultipartUpload. --- s3proxy/internal/router/object.go | 52 ++++++++++++++++++++----------- 1 file changed, 34 insertions(+), 18 deletions(-) diff --git a/s3proxy/internal/router/object.go b/s3proxy/internal/router/object.go index d7be9d6..abb1a24 100644 --- a/s3proxy/internal/router/object.go +++ b/s3proxy/internal/router/object.go @@ -195,6 +195,37 @@ func (o object) put(w http.ResponseWriter, r *http.Request) { log := o.log.With("request_id", requestID) log.Debug("putObject", "key", o.key, "bucket", o.bucket) + // Detach from the request cancellation so a client disconnect does not abort a + // partially-uploaded PutObject, but cap the total duration so a hung upstream + // cannot produce a zombie request. + ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), s3OperationTimeout) + defer cancel() + + output, err := o.encryptAndUpload(ctx, log) + if err != nil { + code := parseErrorCode(err) + if code != 0 { + http.Error(w, err.Error(), code) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + setPutObjectHeaders(w, output) + + w.WriteHeader(http.StatusOK) + if _, err := w.Write(nil); err != nil { + log.Error("PutObject sending response", "error", err) + } +} + +// encryptAndUpload encrypts the object body with a fresh per-object DEK, records the +// wrapped DEK and KEK-version metadata, and stores the ciphertext upstream as a single +// PutObject. It releases the large plaintext and ciphertext buffers as soon as they are +// no longer needed. The caller supplies a context (detached from the request and bounded +// by s3OperationTimeout) so a client disconnect cannot abort an in-flight store. +func (o object) encryptAndUpload(ctx context.Context, log *slog.Logger) (*s3.PutObjectOutput, error) { kekVersion, kek := o.keks.Current() encryptStart := time.Now() ciphertext, encryptedDEK, err := cryptoutil.Encrypt(o.data, kek) @@ -205,15 +236,11 @@ func (o object) put(w http.ResponseWriter, r *http.Request) { releaseLargeBuffer(&o.data) if err != nil { log.Error("PutObject", "error", err) - http.Error(w, err.Error(), http.StatusInternalServerError) - return + return nil, err } o.metadata[config.GetDekTagName()] = hex.EncodeToString(encryptedDEK) o.metadata[config.GetKEKVersionTagName()] = kekVersion - ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), s3OperationTimeout) - defer cancel() - output, err := o.client.PutObject(ctx, o.bucket, o.key, o.tags, o.contentType, o.objectLockLegalHoldStatus, o.objectLockMode, o.sseCustomerAlgorithm, o.sseCustomerKey, o.sseCustomerKeyMD5, o.objectLockRetainUntilDate, o.metadata, ciphertext) // Release the ciphertext buffer as soon as the upstream call returns, win // or lose: the bytes are either persisted upstream or we are about to fail @@ -222,21 +249,10 @@ func (o object) put(w http.ResponseWriter, r *http.Request) { if err != nil { log.Error("PutObject sending request to S3", "error", err) o.incUpstreamError() - code := parseErrorCode(err) - if code != 0 { - http.Error(w, err.Error(), code) - return - } - http.Error(w, err.Error(), http.StatusInternalServerError) - return + return nil, err } - setPutObjectHeaders(w, output) - - w.WriteHeader(http.StatusOK) - if _, err := w.Write(nil); err != nil { - log.Error("PutObject sending response", "error", err) - } + return output, nil } func setPutObjectHeaders(w http.ResponseWriter, output *s3.PutObjectOutput) { From b6e8557ee440016b980195d054d679cf4f8b1dbb Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Wed, 24 Jun 2026 16:25:39 +0200 Subject: [PATCH 2/9] feat(multipart): add disk-buffer session manager (no wiring yet) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Manager that buffers each part to local disk and assembles them into a single plaintext buffer for the existing encrypt→PutObject path. Not yet wired into the router or main. - Create/WritePart/Assemble/Abort over a baseDir + in-memory session table - per-part MD5 hex ETags; running object-size cap enforced during the streamed copy (LimitReader, never trusts a client-supplied length) - Cleanup goroutine: TTL eviction + orphan-dir sweep (reclaims dirs left by a restart, since the session table is not persisted) - nil-safe Metrics interface; package does not import monitoring to keep the dependency direction clean (concrete impl lands in Phase 5) Single-instance / session-affinity component by design. Tests cover create, out-of-order parts, re-upload overwrite, cap enforcement, TTL eviction, orphan sweep, concurrency, and metrics. --- s3proxy/internal/multipart/manager.go | 447 +++++++++++++++++++++ s3proxy/internal/multipart/manager_test.go | 346 ++++++++++++++++ 2 files changed, 793 insertions(+) create mode 100644 s3proxy/internal/multipart/manager.go create mode 100644 s3proxy/internal/multipart/manager_test.go diff --git a/s3proxy/internal/multipart/manager.go b/s3proxy/internal/multipart/manager.go new file mode 100644 index 0000000..5384653 --- /dev/null +++ b/s3proxy/internal/multipart/manager.go @@ -0,0 +1,447 @@ +/* +Copyright (c) Intrinsec 2026 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +/* +Package multipart implements a disk-backed buffer for S3 multipart uploads. + +s3proxy's crypto path (AES-256-GCM-SIV envelope) is two-pass and cannot stream, +so multipart uploads cannot be encrypted part-by-part. Instead the Manager +accepts the full multipart protocol, buffers each part to local disk, and on +completion concatenates the parts into a single plaintext buffer that the caller +encrypts and stores upstream as one PutObject. + +Because the buffers are node-local and the in-memory session table is not +persisted, the Manager is a single-instance, session-affinity component: every +UploadPart for a given upload must reach the process that created it, and a +restart drops in-flight uploads (their on-disk directories are reclaimed by the +orphan sweep). See Cleanup. +*/ +package multipart + +import ( + "context" + "crypto/md5" + "encoding/hex" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "sync" + "time" + + "github.com/google/uuid" +) + +// maxPartNumber is the S3 ceiling on the part number in an UploadPart request. +const maxPartNumber = 10000 + +// cleanupInterval bounds how often the Cleanup goroutine evicts expired sessions +// and sweeps orphaned directories. It is clamped to the configured TTL so a short +// TTL still gets timely collection. +const cleanupInterval = 5 * time.Minute + +var ( + // ErrUploadNotFound is returned when an uploadID has no live session (never + // created, already completed/aborted, or lost to a restart). + ErrUploadNotFound = errors.New("multipart upload not found") + // ErrInvalidPart is returned for an out-of-range part number or a part that is + // referenced at completion but was never uploaded. + ErrInvalidPart = errors.New("invalid multipart part") + // ErrObjectTooLarge is returned when a part write or the assembled object would + // exceed the configured maximum object size. + ErrObjectTooLarge = errors.New("multipart object exceeds maximum size") +) + +// Metadata captures the request fields recorded at CreateMultipartUpload time and +// replayed at completion to build the single upstream PutObject. +type Metadata struct { + ContentType string + Tags string + UserMetadata map[string]string + ObjectLockLegalHoldStatus string + ObjectLockMode string + ObjectLockRetainUntilDate time.Time + SSECustomerAlgorithm string + SSECustomerKey string + SSECustomerKeyMD5 string +} + +// Metrics is the subset of monitoring counters the Manager updates. It is +// satisfied by the process Metrics bundle (wired in a later phase) and may be nil, +// in which case all updates are no-ops. +type Metrics interface { + AddUploadsActive(delta float64) + IncParts() + AddBufferBytes(delta float64) + ObserveAssembleSeconds(seconds float64) +} + +// partInfo records the on-disk size and synthesized ETag of a buffered part. +type partInfo struct { + size int64 + etag string +} + +// session holds the state of a single in-flight multipart upload. Its own mutex +// guards parts so concurrent UploadPart writes to different parts of the same +// upload do not race; the Manager mutex guards the sessions map. +type session struct { + uploadID string + bucket string + key string + dir string + createdAt time.Time + meta Metadata + + mu sync.Mutex + parts map[int]partInfo +} + +// Manager owns the on-disk buffer directory and the in-memory table of in-flight +// multipart uploads. +type Manager struct { + baseDir string + maxObjectSize int64 + ttl time.Duration + log *slog.Logger + metrics Metrics + + // now is the clock used for TTL eviction; overridable in tests. + now func() time.Time + + mu sync.Mutex + sessions map[string]*session +} + +// NewManager creates the buffer directory and returns a ready Manager. log must be +// non-nil; metrics may be nil. +func NewManager(baseDir string, maxObjectSize int64, ttl time.Duration, log *slog.Logger, metrics Metrics) (*Manager, error) { + if err := os.MkdirAll(baseDir, 0o700); err != nil { + return nil, fmt.Errorf("creating multipart buffer dir %q: %w", baseDir, err) + } + return &Manager{ + baseDir: baseDir, + maxObjectSize: maxObjectSize, + ttl: ttl, + log: log, + metrics: metrics, + now: time.Now, + sessions: make(map[string]*session), + }, nil +} + +// Create registers a new upload, creates its on-disk directory and returns the +// generated uploadID. +func (m *Manager) Create(bucket, key string, meta Metadata) (string, error) { + uploadID := uuid.NewString() + dir := filepath.Join(m.baseDir, uploadID) + if err := os.MkdirAll(dir, 0o700); err != nil { + return "", fmt.Errorf("creating upload dir: %w", err) + } + + s := &session{ + uploadID: uploadID, + bucket: bucket, + key: key, + dir: dir, + createdAt: m.now(), + meta: meta, + parts: make(map[int]partInfo), + } + + m.mu.Lock() + m.sessions[uploadID] = s + m.mu.Unlock() + + m.addUploadsActive(1) + m.log.Debug("multipart upload created", "upload_id", uploadID, "bucket", bucket, "key", key) + return uploadID, nil +} + +// WritePart streams body to part- (overwriting any previous upload of +// the same part), enforces the running object-size cap, and returns the part's +// hex MD5 ETag. The cap is checked against the sum of the other parts so a +// re-upload of an existing part is sized correctly. +func (m *Manager) WritePart(uploadID string, partNumber int, body io.Reader) (string, error) { + if partNumber < 1 || partNumber > maxPartNumber { + return "", fmt.Errorf("%w: part number %d out of range [1,%d]", ErrInvalidPart, partNumber, maxPartNumber) + } + + s, err := m.lookup(uploadID) + if err != nil { + return "", err + } + + s.mu.Lock() + defer s.mu.Unlock() + + // Budget for this part is the cap minus everything already buffered for the + // other parts; a re-upload of partNumber does not count against itself. + var others int64 + for n, p := range s.parts { + if n != partNumber { + others += p.size + } + } + budget := m.maxObjectSize - others + + partPath := filepath.Join(s.dir, fmt.Sprintf("part-%d", partNumber)) + // #nosec G304 -- partPath is derived from the manager-owned baseDir, a UUID and a validated part number. + f, err := os.OpenFile(partPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) + if err != nil { + return "", fmt.Errorf("opening part file: %w", err) + } + + // #nosec G401 -- MD5 only synthesizes the S3 part ETag, not for security. + h := md5.New() + // Read one byte past the budget so we can detect an over-cap part without + // trusting any client-provided length. + written, copyErr := io.Copy(io.MultiWriter(f, h), io.LimitReader(body, budget+1)) + closeErr := f.Close() + if copyErr != nil { + _ = os.Remove(partPath) + return "", fmt.Errorf("writing part: %w", copyErr) + } + if closeErr != nil { + _ = os.Remove(partPath) + return "", fmt.Errorf("closing part file: %w", closeErr) + } + if written > budget { + _ = os.Remove(partPath) + return "", fmt.Errorf("%w: part %d would push the object past %d bytes", ErrObjectTooLarge, partNumber, m.maxObjectSize) + } + + etag := hex.EncodeToString(h.Sum(nil)) + + prev, existed := s.parts[partNumber] + s.parts[partNumber] = partInfo{size: written, etag: etag} + + delta := written + if existed { + delta -= prev.size + } else { + m.incParts() + } + m.addBufferBytes(delta) + + return etag, nil +} + +// Assemble validates that every requested part exists, preallocates the total +// size, reads the parts in the given order into a single buffer, and returns it +// with the captured metadata. The order of parts is the caller's responsibility +// (it mirrors the client's CompleteMultipartUpload part list). +func (m *Manager) Assemble(uploadID string, parts []int) ([]byte, Metadata, error) { + s, err := m.lookup(uploadID) + if err != nil { + return nil, Metadata{}, err + } + + start := m.now() + + s.mu.Lock() + defer s.mu.Unlock() + + var total int64 + for _, n := range parts { + p, ok := s.parts[n] + if !ok { + return nil, Metadata{}, fmt.Errorf("%w: part %d was not uploaded", ErrInvalidPart, n) + } + total += p.size + } + if total > m.maxObjectSize { + return nil, Metadata{}, fmt.Errorf("%w: assembled size %d exceeds %d", ErrObjectTooLarge, total, m.maxObjectSize) + } + + data := make([]byte, total) + var off int64 + for _, n := range parts { + size := s.parts[n].size + partPath := filepath.Join(s.dir, fmt.Sprintf("part-%d", n)) + // #nosec G304 -- partPath is derived from the manager-owned baseDir, a UUID and a validated part number. + f, err := os.Open(partPath) + if err != nil { + return nil, Metadata{}, fmt.Errorf("opening part %d: %w", n, err) + } + if _, err := io.ReadFull(f, data[off:off+size]); err != nil { + _ = f.Close() + return nil, Metadata{}, fmt.Errorf("reading part %d: %w", n, err) + } + if err := f.Close(); err != nil { + return nil, Metadata{}, fmt.Errorf("closing part %d: %w", n, err) + } + off += size + } + + m.observeAssemble(m.now().Sub(start)) + return data, s.meta, nil +} + +// Abort drops the session and removes its on-disk directory. It is idempotent: +// aborting an unknown upload is a no-op so it can double as post-completion +// cleanup. +func (m *Manager) Abort(uploadID string) error { + m.mu.Lock() + s, ok := m.sessions[uploadID] + if ok { + delete(m.sessions, uploadID) + } + m.mu.Unlock() + if !ok { + return nil + } + return m.discard(s) +} + +// discard removes a session's directory and reverses its metric contributions. +func (m *Manager) discard(s *session) error { + s.mu.Lock() + bytesHeld := sessionBytes(s) + dir := s.dir + s.mu.Unlock() + + err := os.RemoveAll(dir) + m.addUploadsActive(-1) + m.addBufferBytes(-bytesHeld) + if err != nil { + return fmt.Errorf("removing upload dir: %w", err) + } + return nil +} + +// Cleanup runs until ctx is cancelled, periodically evicting expired sessions and +// sweeping orphaned directories. It performs one sweep immediately so directories +// left by a previous run are reclaimed at startup. +func (m *Manager) Cleanup(ctx context.Context) { + m.sweepOrphans() + + interval := cleanupInterval + if m.ttl > 0 && m.ttl < interval { + interval = m.ttl + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + m.evictExpired(m.now()) + m.sweepOrphans() + } + } +} + +// evictExpired removes sessions whose age exceeds the TTL relative to now. +func (m *Manager) evictExpired(now time.Time) { + m.mu.Lock() + var expired []*session + for id, s := range m.sessions { + if now.Sub(s.createdAt) > m.ttl { + expired = append(expired, s) + delete(m.sessions, id) + } + } + m.mu.Unlock() + + for _, s := range expired { + if err := m.discard(s); err != nil { + m.log.Warn("evicting expired multipart upload", "upload_id", s.uploadID, "error", err) + continue + } + m.log.Info("evicted expired multipart upload", "upload_id", s.uploadID) + } +} + +// sweepOrphans removes directories under baseDir that no live session references +// and that are older than the TTL. After a restart the session table is empty, so +// every leftover directory is an orphan; the age guard avoids racing a concurrent +// Create. +func (m *Manager) sweepOrphans() { + entries, err := os.ReadDir(m.baseDir) + if err != nil { + m.log.Warn("reading multipart buffer dir", "dir", m.baseDir, "error", err) + return + } + + m.mu.Lock() + known := make(map[string]struct{}, len(m.sessions)) + for id := range m.sessions { + known[id] = struct{}{} + } + m.mu.Unlock() + + now := m.now() + for _, e := range entries { + if !e.IsDir() { + continue + } + if _, ok := known[e.Name()]; ok { + continue + } + info, err := e.Info() + if err != nil { + continue + } + if now.Sub(info.ModTime()) <= m.ttl { + continue + } + path := filepath.Join(m.baseDir, e.Name()) + if err := os.RemoveAll(path); err != nil { + m.log.Warn("removing orphan multipart dir", "dir", path, "error", err) + continue + } + m.log.Info("removed orphan multipart dir", "dir", path) + } +} + +// lookup returns the session for uploadID or ErrUploadNotFound. +func (m *Manager) lookup(uploadID string) (*session, error) { + m.mu.Lock() + defer m.mu.Unlock() + s, ok := m.sessions[uploadID] + if !ok { + return nil, fmt.Errorf("%w: %s", ErrUploadNotFound, uploadID) + } + return s, nil +} + +// sessionBytes sums the buffered bytes of a session. Caller holds s.mu. +func sessionBytes(s *session) int64 { + var total int64 + for _, p := range s.parts { + total += p.size + } + return total +} + +func (m *Manager) addUploadsActive(delta float64) { + if m.metrics != nil { + m.metrics.AddUploadsActive(delta) + } +} + +func (m *Manager) incParts() { + if m.metrics != nil { + m.metrics.IncParts() + } +} + +func (m *Manager) addBufferBytes(delta int64) { + if m.metrics != nil { + m.metrics.AddBufferBytes(float64(delta)) + } +} + +func (m *Manager) observeAssemble(d time.Duration) { + if m.metrics != nil { + m.metrics.ObserveAssembleSeconds(d.Seconds()) + } +} diff --git a/s3proxy/internal/multipart/manager_test.go b/s3proxy/internal/multipart/manager_test.go new file mode 100644 index 0000000..79322b5 --- /dev/null +++ b/s3proxy/internal/multipart/manager_test.go @@ -0,0 +1,346 @@ +/* +Copyright (c) Intrinsec 2026 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package multipart + +import ( + "bytes" + "crypto/md5" //nolint:gosec // test asserts the manager's ETag matches the MD5 of the part. + "encoding/hex" + "errors" + "io" + "log/slog" + "os" + "path/filepath" + "sync" + "testing" + "time" +) + +const testCap = 1 << 20 // 1 MiB + +func newTestManager(t *testing.T) *Manager { + t.Helper() + m, err := NewManager(t.TempDir(), testCap, time.Hour, slog.New(slog.NewTextHandler(io.Discard, nil)), nil) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + return m +} + +func md5Hex(b []byte) string { + sum := md5.Sum(b) //nolint:gosec // see import note. + return hex.EncodeToString(sum[:]) +} + +func TestCreateRegistersSessionAndDir(t *testing.T) { + m := newTestManager(t) + + id, err := m.Create("bucket", "key", Metadata{ContentType: "text/plain"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + if id == "" { + t.Fatal("Create returned empty uploadID") + } + + if _, err := os.Stat(filepath.Join(m.baseDir, id)); err != nil { + t.Fatalf("upload dir missing: %v", err) + } + if _, err := m.lookup(id); err != nil { + t.Fatalf("session not registered: %v", err) + } +} + +func TestWritePartETagAndAssemble(t *testing.T) { + m := newTestManager(t) + id, err := m.Create("bucket", "key", Metadata{ContentType: "application/octet-stream"}) + if err != nil { + t.Fatalf("Create: %v", err) + } + + part1 := bytes.Repeat([]byte("a"), 1000) + part2 := bytes.Repeat([]byte("b"), 500) + + // Upload out of order: part 2 first, then part 1. + etag2, err := m.WritePart(id, 2, bytes.NewReader(part2)) + if err != nil { + t.Fatalf("WritePart 2: %v", err) + } + if etag2 != md5Hex(part2) { + t.Fatalf("etag2 = %s, want %s", etag2, md5Hex(part2)) + } + etag1, err := m.WritePart(id, 1, bytes.NewReader(part1)) + if err != nil { + t.Fatalf("WritePart 1: %v", err) + } + if etag1 != md5Hex(part1) { + t.Fatalf("etag1 = %s, want %s", etag1, md5Hex(part1)) + } + + data, meta, err := m.Assemble(id, []int{1, 2}) + if err != nil { + t.Fatalf("Assemble: %v", err) + } + want := append(append([]byte{}, part1...), part2...) + if !bytes.Equal(data, want) { + t.Fatalf("assembled data mismatch: got %d bytes, want %d", len(data), len(want)) + } + if meta.ContentType != "application/octet-stream" { + t.Fatalf("metadata not preserved: %+v", meta) + } +} + +func TestWritePartReupload(t *testing.T) { + m := newTestManager(t) + id, _ := m.Create("bucket", "key", Metadata{}) + + if _, err := m.WritePart(id, 1, bytes.NewReader(bytes.Repeat([]byte("x"), 100))); err != nil { + t.Fatalf("first WritePart: %v", err) + } + newBody := bytes.Repeat([]byte("y"), 250) + etag, err := m.WritePart(id, 1, bytes.NewReader(newBody)) + if err != nil { + t.Fatalf("re-upload WritePart: %v", err) + } + if etag != md5Hex(newBody) { + t.Fatalf("re-upload etag mismatch") + } + + data, _, err := m.Assemble(id, []int{1}) + if err != nil { + t.Fatalf("Assemble: %v", err) + } + if !bytes.Equal(data, newBody) { + t.Fatalf("re-upload did not overwrite: got %d bytes", len(data)) + } +} + +func TestWritePartCapEnforced(t *testing.T) { + m, err := NewManager(t.TempDir(), 1000, time.Hour, slog.New(slog.NewTextHandler(io.Discard, nil)), nil) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + id, _ := m.Create("bucket", "key", Metadata{}) + + if _, err := m.WritePart(id, 1, bytes.NewReader(bytes.Repeat([]byte("a"), 600))); err != nil { + t.Fatalf("WritePart 1: %v", err) + } + // Second part pushes total to 1100 > 1000. + _, err = m.WritePart(id, 2, bytes.NewReader(bytes.Repeat([]byte("b"), 500))) + if !errors.Is(err, ErrObjectTooLarge) { + t.Fatalf("want ErrObjectTooLarge, got %v", err) + } + // The rejected part must not be left on disk. + if _, statErr := os.Stat(filepath.Join(m.baseDir, id, "part-2")); !os.IsNotExist(statErr) { + t.Fatalf("rejected part file still present: %v", statErr) + } +} + +func TestWritePartInvalidNumber(t *testing.T) { + m := newTestManager(t) + id, _ := m.Create("bucket", "key", Metadata{}) + + for _, n := range []int{0, -1, maxPartNumber + 1} { + if _, err := m.WritePart(id, n, bytes.NewReader([]byte("x"))); !errors.Is(err, ErrInvalidPart) { + t.Fatalf("part %d: want ErrInvalidPart, got %v", n, err) + } + } +} + +func TestWritePartUnknownUpload(t *testing.T) { + m := newTestManager(t) + if _, err := m.WritePart("nope", 1, bytes.NewReader([]byte("x"))); !errors.Is(err, ErrUploadNotFound) { + t.Fatalf("want ErrUploadNotFound, got %v", err) + } +} + +func TestAssembleMissingPart(t *testing.T) { + m := newTestManager(t) + id, _ := m.Create("bucket", "key", Metadata{}) + if _, err := m.WritePart(id, 1, bytes.NewReader([]byte("x"))); err != nil { + t.Fatalf("WritePart: %v", err) + } + if _, _, err := m.Assemble(id, []int{1, 2}); !errors.Is(err, ErrInvalidPart) { + t.Fatalf("want ErrInvalidPart, got %v", err) + } +} + +func TestAbortRemovesSessionAndDir(t *testing.T) { + m := newTestManager(t) + id, _ := m.Create("bucket", "key", Metadata{}) + if _, err := m.WritePart(id, 1, bytes.NewReader([]byte("data"))); err != nil { + t.Fatalf("WritePart: %v", err) + } + dir := filepath.Join(m.baseDir, id) + + if err := m.Abort(id); err != nil { + t.Fatalf("Abort: %v", err) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("dir not removed: %v", err) + } + if _, err := m.lookup(id); !errors.Is(err, ErrUploadNotFound) { + t.Fatalf("session still present: %v", err) + } + // Idempotent: aborting again is a no-op. + if err := m.Abort(id); err != nil { + t.Fatalf("second Abort: %v", err) + } +} + +func TestEvictExpired(t *testing.T) { + ttl := time.Minute + m, err := NewManager(t.TempDir(), testCap, ttl, slog.New(slog.NewTextHandler(io.Discard, nil)), nil) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + id, _ := m.Create("bucket", "key", Metadata{}) + dir := filepath.Join(m.baseDir, id) + + // Not yet expired. + m.evictExpired(m.now()) + if _, err := m.lookup(id); err != nil { + t.Fatalf("session evicted too early: %v", err) + } + + // Past the TTL. + m.evictExpired(time.Now().Add(2 * ttl)) + if _, err := m.lookup(id); !errors.Is(err, ErrUploadNotFound) { + t.Fatalf("expired session not evicted: %v", err) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("expired dir not removed: %v", err) + } +} + +func TestSweepOrphans(t *testing.T) { + ttl := time.Minute + m, err := NewManager(t.TempDir(), testCap, ttl, slog.New(slog.NewTextHandler(io.Discard, nil)), nil) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + + // A live session's dir must survive the sweep. + liveID, _ := m.Create("bucket", "key", Metadata{}) + + // A stale orphan dir (unknown to the table, older than the TTL) must be removed. + orphan := filepath.Join(m.baseDir, "orphan-upload") + if err := os.MkdirAll(orphan, 0o700); err != nil { + t.Fatalf("mkdir orphan: %v", err) + } + old := time.Now().Add(-2 * ttl) + if err := os.Chtimes(orphan, old, old); err != nil { + t.Fatalf("chtimes: %v", err) + } + + // A fresh orphan (younger than the TTL) must be left alone. + freshOrphan := filepath.Join(m.baseDir, "fresh-orphan") + if err := os.MkdirAll(freshOrphan, 0o700); err != nil { + t.Fatalf("mkdir fresh orphan: %v", err) + } + + m.sweepOrphans() + + if _, err := os.Stat(orphan); !os.IsNotExist(err) { + t.Fatalf("stale orphan not swept: %v", err) + } + if _, err := os.Stat(freshOrphan); err != nil { + t.Fatalf("fresh orphan wrongly swept: %v", err) + } + if _, err := os.Stat(filepath.Join(m.baseDir, liveID)); err != nil { + t.Fatalf("live session dir wrongly swept: %v", err) + } +} + +func TestConcurrentWritePartsDifferentParts(t *testing.T) { + m := newTestManager(t) + id, _ := m.Create("bucket", "key", Metadata{}) + + var wg sync.WaitGroup + for n := 1; n <= 8; n++ { + wg.Add(1) + go func(part int) { + defer wg.Done() + body := bytes.Repeat([]byte{byte('0' + part)}, 100) + if _, err := m.WritePart(id, part, bytes.NewReader(body)); err != nil { + t.Errorf("WritePart %d: %v", part, err) + } + }(n) + } + wg.Wait() + + data, _, err := m.Assemble(id, []int{1, 2, 3, 4, 5, 6, 7, 8}) + if err != nil { + t.Fatalf("Assemble: %v", err) + } + if len(data) != 800 { + t.Fatalf("assembled size = %d, want 800", len(data)) + } +} + +func TestMetricsInvoked(t *testing.T) { + rec := &recordingMetrics{} + m, err := NewManager(t.TempDir(), testCap, time.Hour, slog.New(slog.NewTextHandler(io.Discard, nil)), rec) + if err != nil { + t.Fatalf("NewManager: %v", err) + } + id, _ := m.Create("bucket", "key", Metadata{}) + if _, err := m.WritePart(id, 1, bytes.NewReader(bytes.Repeat([]byte("a"), 100))); err != nil { + t.Fatalf("WritePart: %v", err) + } + if _, _, err := m.Assemble(id, []int{1}); err != nil { + t.Fatalf("Assemble: %v", err) + } + if err := m.Abort(id); err != nil { + t.Fatalf("Abort: %v", err) + } + + if rec.uploadsActive != 0 { + t.Errorf("uploadsActive = %v, want 0 (created then aborted)", rec.uploadsActive) + } + if rec.parts != 1 { + t.Errorf("parts = %d, want 1", rec.parts) + } + if rec.bufferBytes != 0 { + t.Errorf("bufferBytes = %v, want 0 after abort", rec.bufferBytes) + } + if rec.assembleObservations != 1 { + t.Errorf("assembleObservations = %d, want 1", rec.assembleObservations) + } +} + +type recordingMetrics struct { + mu sync.Mutex + uploadsActive float64 + parts int + bufferBytes float64 + assembleObservations int +} + +func (r *recordingMetrics) AddUploadsActive(delta float64) { + r.mu.Lock() + defer r.mu.Unlock() + r.uploadsActive += delta +} + +func (r *recordingMetrics) IncParts() { + r.mu.Lock() + defer r.mu.Unlock() + r.parts++ +} + +func (r *recordingMetrics) AddBufferBytes(delta float64) { + r.mu.Lock() + defer r.mu.Unlock() + r.bufferBytes += delta +} + +func (r *recordingMetrics) ObserveAssembleSeconds(float64) { + r.mu.Lock() + defer r.mu.Unlock() + r.assembleObservations++ +} From 584aaad61c461c5998010a50bdef42f76a6403e7 Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Wed, 24 Jun 2026 16:45:47 +0200 Subject: [PATCH 3/9] feat(config): add multipart buffer-mode knobs (not yet wired) Expose three S3PROXY_MULTIPART_* settings for the upcoming disk-buffer multipart mode; inert until the handlers/main wiring read them. - MultipartBufferDir <- S3PROXY_MULTIPART_BUFFER_DIR ("" = disabled) - MultipartMaxSize <- S3PROXY_MULTIPART_MAX_SIZE (default/clamp 5 GiB) - MultipartTTL <- S3PROXY_MULTIPART_TTL (default 24h) Includes matching GetX shims and config_test.go for defaults/clamping. --- s3proxy/internal/config/config.go | 51 +++++++++++++++++ s3proxy/internal/config/config_test.go | 78 ++++++++++++++++++++++++++ 2 files changed, 129 insertions(+) create mode 100644 s3proxy/internal/config/config_test.go diff --git a/s3proxy/internal/config/config.go b/s3proxy/internal/config/config.go index 7c41a7d..53f3fca 100644 --- a/s3proxy/internal/config/config.go +++ b/s3proxy/internal/config/config.go @@ -11,11 +11,16 @@ import ( "errors" "fmt" "strings" + "time" "github.com/knadh/koanf/providers/env" "github.com/knadh/koanf/v2" ) +// DefaultMultipartTTL is how long an idle buffered multipart session is retained +// before the Cleanup sweep evicts it and removes its on-disk parts. +const DefaultMultipartTTL = 24 * time.Hour + // Config holds every s3proxy runtime setting loaded from the environment. type Config struct { k *koanf.Koanf @@ -95,6 +100,43 @@ func (c *Config) MaxPutBodySize() int64 { return v } +// MultipartBufferDir returns the local directory used to buffer multipart upload +// parts (S3PROXY_MULTIPART_BUFFER_DIR). An empty string (the default) leaves +// buffer-mode multipart disabled. +func (c *Config) MultipartBufferDir() string { + return c.k.String("s3proxy.multipart.buffer.dir") +} + +// MultipartMaxSize returns the maximum assembled object size in bytes for a +// buffered multipart upload (S3PROXY_MULTIPART_MAX_SIZE). Falls back to +// MaxObjectSize when unset, non-positive, or larger than MaxObjectSize. +func (c *Config) MultipartMaxSize() int64 { + const key = "s3proxy.multipart.max.size" + if !c.k.Exists(key) { + return MaxObjectSize + } + v := c.k.Int64(key) + if v <= 0 || v > MaxObjectSize { + return MaxObjectSize + } + return v +} + +// MultipartTTL returns how long an idle buffered multipart session is retained +// before eviction (S3PROXY_MULTIPART_TTL). Falls back to DefaultMultipartTTL +// when unset, non-positive, or unparseable. +func (c *Config) MultipartTTL() time.Duration { + const key = "s3proxy.multipart.ttl" + if !c.k.Exists(key) { + return DefaultMultipartTTL + } + v := c.k.Duration(key) + if v <= 0 { + return DefaultMultipartTTL + } + return v +} + // DecryptionFallback reports whether GetObject should retry decryption with an // all-zero KEK when the configured KEK fails (S3PROXY_DECRYPTION_FALLBACK=1). // Intended for one-shot migrations away from objects written without an @@ -157,3 +199,12 @@ func GetInsecure() bool { return Default().Insecure() } // GetDecryptionFallback returns the decryption-fallback flag from the default config. func GetDecryptionFallback() bool { return Default().DecryptionFallback() } + +// GetMultipartBufferDir returns the multipart buffer directory from the default config. +func GetMultipartBufferDir() string { return Default().MultipartBufferDir() } + +// GetMultipartMaxSize returns the multipart max object size from the default config. +func GetMultipartMaxSize() int64 { return Default().MultipartMaxSize() } + +// GetMultipartTTL returns the multipart session TTL from the default config. +func GetMultipartTTL() time.Duration { return Default().MultipartTTL() } diff --git a/s3proxy/internal/config/config_test.go b/s3proxy/internal/config/config_test.go new file mode 100644 index 0000000..fba07b0 --- /dev/null +++ b/s3proxy/internal/config/config_test.go @@ -0,0 +1,78 @@ +/* +Copyright (c) Intrinsec 2026 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package config + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMultipartBufferDir(t *testing.T) { + t.Run("unset disables buffer mode", func(t *testing.T) { + cfg, err := Load() + require.NoError(t, err) + assert.Empty(t, cfg.MultipartBufferDir()) + }) + + t.Run("set returns the directory", func(t *testing.T) { + t.Setenv("S3PROXY_MULTIPART_BUFFER_DIR", "/var/lib/s3proxy/parts") + cfg, err := Load() + require.NoError(t, err) + assert.Equal(t, "/var/lib/s3proxy/parts", cfg.MultipartBufferDir()) + }) +} + +func TestMultipartMaxSize(t *testing.T) { + tests := map[string]struct { + set bool + env string + want int64 + }{ + "unset defaults to MaxObjectSize": {set: false, want: MaxObjectSize}, + "valid value honoured": {set: true, env: "104857600", want: 104857600}, + "at limit honoured": {set: true, env: "5368709120", want: MaxObjectSize}, + "over limit clamps to default": {set: true, env: "5368709121", want: MaxObjectSize}, + "zero falls back to default": {set: true, env: "0", want: MaxObjectSize}, + "negative falls back to default": {set: true, env: "-1", want: MaxObjectSize}, + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + if tc.set { + t.Setenv("S3PROXY_MULTIPART_MAX_SIZE", tc.env) + } + cfg, err := Load() + require.NoError(t, err) + assert.Equal(t, tc.want, cfg.MultipartMaxSize()) + }) + } +} + +func TestMultipartTTL(t *testing.T) { + tests := map[string]struct { + set bool + env string + want time.Duration + }{ + "unset defaults to 24h": {set: false, want: DefaultMultipartTTL}, + "valid duration honoured": {set: true, env: "1h30m", want: 90 * time.Minute}, + "empty falls back to default": {set: true, env: "", want: DefaultMultipartTTL}, + "unparseable falls back": {set: true, env: "nope", want: DefaultMultipartTTL}, + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + if tc.set { + t.Setenv("S3PROXY_MULTIPART_TTL", tc.env) + } + cfg, err := Load() + require.NoError(t, err) + assert.Equal(t, tc.want, cfg.MultipartTTL()) + }) + } +} From 34358256ef34cf6f3be3a5f5b3c390cbb438109f Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Wed, 24 Jun 2026 17:16:32 +0200 Subject: [PATCH 4/9] feat(multipart): wire buffer-mode multipart handlers + router Replace the multipart 501 stubs with buffer-mode handlers when a *multipart.Manager is wired. Create captures object metadata and opens a disk-backed session; UploadPart de-frames aws-chunked bodies and streams each part to disk; Complete assembles the parts, reuses encryptAndUpload to store one AES-256-GCM-SIV PutObject upstream, then drops the buffer; Abort releases the session. getMultipartHandler now follows forward -> buffer -> block precedence. Serve exempts UploadPart from the 256 MiB PutObject body cap while keeping the 5 GiB ContentLength ceiling. New aws-chunked de-framer and S3 multipart XML structs back the handlers. Complete responds 200 only after upstream success and keeps the session on failure so the client can retry, matching the single-shot ack order. --- s3proxy/cmd/main.go | 4 +- s3proxy/internal/router/handler_multipart.go | 264 +++++++++++++++++++ s3proxy/internal/router/multipart_test.go | 212 +++++++++++++++ s3proxy/internal/router/multipart_xml.go | 153 +++++++++++ s3proxy/internal/router/router.go | 75 ++++-- 5 files changed, 681 insertions(+), 27 deletions(-) create mode 100644 s3proxy/internal/router/handler_multipart.go create mode 100644 s3proxy/internal/router/multipart_test.go create mode 100644 s3proxy/internal/router/multipart_xml.go diff --git a/s3proxy/cmd/main.go b/s3proxy/cmd/main.go index 169fb95..4354d60 100644 --- a/s3proxy/cmd/main.go +++ b/s3proxy/cmd/main.go @@ -129,7 +129,9 @@ func runServer(flags cmdFlags, cfg *config.Config, log *slog.Logger) error { metrics := monitoring.New() - routerInstance, err := router.New(context.Background(), flags.region, flags.forwardMultipartReqs, log, metrics) + // Buffer-mode multipart (the *multipart.Manager) is wired in a later phase; nil + // keeps multipart blocked by default unless --allow-multipart forwards it. + routerInstance, err := router.New(context.Background(), flags.region, flags.forwardMultipartReqs, nil, log, metrics) if err != nil { return fmt.Errorf("creating router: %w", err) } diff --git a/s3proxy/internal/router/handler_multipart.go b/s3proxy/internal/router/handler_multipart.go new file mode 100644 index 0000000..ae12f7f --- /dev/null +++ b/s3proxy/internal/router/handler_multipart.go @@ -0,0 +1,264 @@ +/* +Copyright (c) Intrinsec 2026 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package router + +import ( + "context" + "encoding/xml" + "errors" + "fmt" + "io" + "net/http" + "strconv" + "strings" + + "github.com/google/uuid" + "github.com/intrinsec/s3proxy/internal/multipart" +) + +// maxCompleteBodySize caps the CompleteMultipartUpload request body. The body is a +// part list (≤10000 entries of a part number + ETag), so a few MiB is ample and the +// limit guards against a hostile or malformed request forcing an unbounded read. +const maxCompleteBodySize = 8 << 20 // 8 MiB + +// handleCreateMultipartUpload (buffer mode) records the request metadata, opens a +// disk-backed session and returns the proxy-generated uploadID. +func (r Router) handleCreateMultipartUpload(key, bucket string) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + r.log.Debug("intercepting CreateMultipartUpload (buffer mode)", "path", req.URL.Path, "method", req.Method, "host", req.Host) + + retentionTime, err := parseRetentionTime(req.Header.Get("x-amz-object-lock-retain-until-date")) + if err != nil { + r.log.Error("CreateMultipartUpload parsing lock retention time", "error", err) + r.incError("multipart_create") + http.Error(w, fmt.Sprintf("parsing x-amz-object-lock-retain-until-date: %s", err.Error()), http.StatusBadRequest) + return + } + + meta := multipart.Metadata{ + ContentType: req.Header.Get("Content-Type"), + Tags: req.Header.Get("x-amz-tagging"), + UserMetadata: getMetadataHeaders(req.Header), + ObjectLockLegalHoldStatus: req.Header.Get("x-amz-object-lock-legal-hold"), + ObjectLockMode: req.Header.Get("x-amz-object-lock-mode"), + ObjectLockRetainUntilDate: retentionTime, + SSECustomerAlgorithm: req.Header.Get("x-amz-server-side-encryption-customer-algorithm"), + SSECustomerKey: req.Header.Get("x-amz-server-side-encryption-customer-key"), + SSECustomerKeyMD5: req.Header.Get("x-amz-server-side-encryption-customer-key-MD5"), + } + + uploadID, err := r.multipart.Create(bucket, key, meta) + if err != nil { + r.log.Error("CreateMultipartUpload", "error", err) + r.incError("multipart_create") + http.Error(w, "failed to create multipart upload", http.StatusInternalServerError) + return + } + + r.writeXML(w, http.StatusOK, InitiateMultipartUploadResult{ + Bucket: bucket, + Key: key, + UploadID: uploadID, + }) + } +} + +// handleUploadPart (buffer mode) streams one part to disk, de-framing aws-chunked +// bodies first, and returns the synthesized per-part ETag. +func (r Router) handleUploadPart(key, bucket string) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + r.log.Debug("intercepting UploadPart (buffer mode)", "path", req.URL.Path, "method", req.Method, "host", req.Host) + + query := req.URL.Query() + uploadID := query.Get("uploadId") + partNumber, err := strconv.Atoi(query.Get("partNumber")) + if err != nil { + r.log.Warn("UploadPart invalid part number", "partNumber", query.Get("partNumber"), "error", err) + r.incError("multipart_upload_part") + http.Error(w, "invalid partNumber", http.StatusBadRequest) + return + } + + var body io.Reader = req.Body + if isAWSChunked(req.Header) { + body = newAWSChunkedReader(req.Body) + } + + etag, err := r.multipart.WritePart(uploadID, partNumber, body) + if err != nil { + r.writeMultipartError(w, "UploadPart", err) + return + } + + // S3 ETags are quoted; SDKs strip the quotes when collecting parts. + w.Header().Set("ETag", strconv.Quote(etag)) + w.WriteHeader(http.StatusOK) + } +} + +// handleCompleteMultipartUpload (buffer mode) assembles the buffered parts into one +// plaintext object, encrypts it and stores it upstream as a single PutObject, then +// drops the buffer. It responds 200 only after the upstream store succeeds, mirroring +// the single-shot PutObject ack ordering: an UploadPart ack means "part buffered", +// not "object stored". On upstream failure the session is kept so the client can +// retry Complete. +func (r Router) handleCompleteMultipartUpload(client s3Client, key, bucket string) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + r.log.Debug("intercepting CompleteMultipartUpload (buffer mode)", "path", req.URL.Path, "method", req.Method, "host", req.Host) + + uploadID := req.URL.Query().Get("uploadId") + + rawBody, err := io.ReadAll(io.LimitReader(req.Body, maxCompleteBodySize+1)) + if err != nil { + r.log.Error("CompleteMultipartUpload reading body", "error", err) + r.incError("multipart_complete") + http.Error(w, "failed to read request body", http.StatusInternalServerError) + return + } + if int64(len(rawBody)) > maxCompleteBodySize { + r.incError("multipart_complete") + http.Error(w, "CompleteMultipartUpload body too large", http.StatusRequestEntityTooLarge) + return + } + + var complete CompleteMultipartUpload + if err := xml.Unmarshal(rawBody, &complete); err != nil { + r.log.Warn("CompleteMultipartUpload malformed body", "error", err) + r.incError("multipart_complete") + http.Error(w, "malformed CompleteMultipartUpload body", http.StatusBadRequest) + return + } + if len(complete.Parts) == 0 { + r.incError("multipart_complete") + http.Error(w, "CompleteMultipartUpload part list is empty", http.StatusBadRequest) + return + } + + parts := make([]int, len(complete.Parts)) + for i, p := range complete.Parts { + parts[i] = p.PartNumber + } + + data, meta, err := r.multipart.Assemble(uploadID, parts) + if err != nil { + r.writeMultipartError(w, "CompleteMultipartUpload", err) + return + } + + obj := object{ + keks: r.keks, + client: client, + key: key, + bucket: bucket, + data: data, + tags: meta.Tags, + contentType: meta.ContentType, + metadata: meta.UserMetadata, + objectLockLegalHoldStatus: meta.ObjectLockLegalHoldStatus, + objectLockMode: meta.ObjectLockMode, + objectLockRetainUntilDate: meta.ObjectLockRetainUntilDate, + sseCustomerAlgorithm: meta.SSECustomerAlgorithm, + sseCustomerKey: meta.SSECustomerKey, + sseCustomerKeyMD5: meta.SSECustomerKeyMD5, + log: r.log, + metrics: r.metrics, + } + + requestID := uuid.New().String() + log := r.log.With("request_id", requestID) + + // Detach from the request context so a client disconnect cannot abort the + // in-flight store, but cap the total duration; same as the single-shot path. + ctx, cancel := context.WithTimeout(context.WithoutCancel(req.Context()), s3OperationTimeout) + defer cancel() + + output, err := obj.encryptAndUpload(ctx, log) + if err != nil { + // Keep the session so the client can retry Complete; never ack a store + // the upstream did not confirm. + if code := parseErrorCode(err); code != 0 { + http.Error(w, err.Error(), code) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Object is durably stored upstream; reclaim the on-disk buffer. + if err := r.multipart.Abort(uploadID); err != nil { + log.Warn("CompleteMultipartUpload cleaning up buffer", "upload_id", uploadID, "error", err) + } + + etag := "" + if output.ETag != nil { + etag = strings.Trim(*output.ETag, "\"") + } + r.writeXML(w, http.StatusOK, CompleteMultipartUploadResult{ + Bucket: bucket, + Key: key, + ETag: strconv.Quote(etag), + }) + } +} + +// handleAbortMultipartUpload (buffer mode) drops the session and its on-disk parts. +// It is idempotent: aborting an unknown upload still returns 204. +func (r Router) handleAbortMultipartUpload() http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + r.log.Debug("intercepting AbortMultipartUpload (buffer mode)", "path", req.URL.Path, "method", req.Method, "host", req.Host) + + uploadID := req.URL.Query().Get("uploadId") + if err := r.multipart.Abort(uploadID); err != nil { + r.log.Error("AbortMultipartUpload", "upload_id", uploadID, "error", err) + r.incError("multipart_abort") + http.Error(w, "failed to abort multipart upload", http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) + } +} + +// writeMultipartError maps a Manager error to an HTTP status and writes it. +func (r Router) writeMultipartError(w http.ResponseWriter, op string, err error) { + switch { + case errors.Is(err, multipart.ErrUploadNotFound): + r.log.Warn(op+" upload not found", "error", err) + r.incError("multipart_not_found") + http.Error(w, err.Error(), http.StatusNotFound) + case errors.Is(err, multipart.ErrInvalidPart): + r.log.Warn(op+" invalid part", "error", err) + r.incError("multipart_invalid_part") + http.Error(w, err.Error(), http.StatusBadRequest) + case errors.Is(err, multipart.ErrObjectTooLarge): + r.log.Warn(op+" object too large", "error", err) + r.incError("multipart_too_large") + http.Error(w, err.Error(), http.StatusRequestEntityTooLarge) + default: + r.log.Error(op+" failed", "error", err) + r.incError("multipart_internal") + http.Error(w, "internal server error", http.StatusInternalServerError) + } +} + +// writeXML marshals v as an S3-style XML response with the declaration prologue. +func (r Router) writeXML(w http.ResponseWriter, status int, v any) { + marshalled, err := xml.Marshal(v) + if err != nil { + r.log.Error("marshalling XML response", "error", err) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/xml") + w.WriteHeader(status) + if _, err := w.Write([]byte(xml.Header)); err != nil { + r.log.Error("writing XML header", "error", err) + return + } + if _, err := w.Write(marshalled); err != nil { + r.log.Error("writing XML body", "error", err) + } +} diff --git a/s3proxy/internal/router/multipart_test.go b/s3proxy/internal/router/multipart_test.go new file mode 100644 index 0000000..24b43e9 --- /dev/null +++ b/s3proxy/internal/router/multipart_test.go @@ -0,0 +1,212 @@ +/* +Copyright (c) Intrinsec 2026 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package router + +import ( + "bytes" + "encoding/hex" + "encoding/xml" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/intrinsec/s3proxy/internal/config" + "github.com/intrinsec/s3proxy/internal/cryptoutil" + "github.com/intrinsec/s3proxy/internal/multipart" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newBufferRouter(t *testing.T, seed string) (Router, *multipart.Manager) { + t.Helper() + keks := newTestKEKs(t, seed) + mgr, err := multipart.NewManager(t.TempDir(), config.MaxObjectSize, time.Hour, testLogger(), nil) + require.NoError(t, err) + return Router{keks: keks, multipart: mgr, log: testLogger()}, mgr +} + +func createUpload(t *testing.T, r Router, client s3Client) string { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/bucket/key?uploads", nil) + rec := httptest.NewRecorder() + r.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code) + + var res InitiateMultipartUploadResult + require.NoError(t, xml.Unmarshal(rec.Body.Bytes(), &res)) + require.NotEmpty(t, res.UploadID) + assert.Equal(t, "bucket", res.Bucket) + assert.Equal(t, "key", res.Key) + return res.UploadID +} + +func uploadPart(t *testing.T, r Router, client s3Client, req *http.Request) { + t.Helper() + rec := httptest.NewRecorder() + r.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code) + assert.NotEmpty(t, rec.Header().Get("ETag")) +} + +func uploadPartReq(uploadID string, n int, data []byte) *http.Request { + url := fmt.Sprintf("/bucket/key?partNumber=%d&uploadId=%s", n, uploadID) + return httptest.NewRequest(http.MethodPut, url, bytes.NewReader(data)) +} + +func completeBody(t *testing.T, parts ...int) []byte { + t.Helper() + c := CompleteMultipartUpload{} + for _, n := range parts { + c.Parts = append(c.Parts, CompletedPart{PartNumber: n, ETag: fmt.Sprintf("etag-%d", n)}) + } + b, err := xml.Marshal(c) + require.NoError(t, err) + return b +} + +func TestMultipartBufferRoundTrip(t *testing.T) { + r, _ := newBufferRouter(t, "multipart seed") + client := &recordingS3Client{} + + uploadID := createUpload(t, r, client) + + part1 := []byte("hello ") + part2 := []byte("buffered multipart world") + uploadPart(t, r, client, uploadPartReq(uploadID, 1, part1)) + uploadPart(t, r, client, uploadPartReq(uploadID, 2, part2)) + + req := httptest.NewRequest(http.MethodPost, "/bucket/key?uploadId="+uploadID, bytes.NewReader(completeBody(t, 1, 2))) + rec := httptest.NewRecorder() + r.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code) + + var res CompleteMultipartUploadResult + require.NoError(t, xml.Unmarshal(rec.Body.Bytes(), &res)) + assert.Equal(t, "bucket", res.Bucket) + assert.Equal(t, "key", res.Key) + + // The upstream object is a single ciphertext carrying the DEK + KEK-version tags. + rawDEK, ok := client.metadata[config.GetDekTagName()] + require.True(t, ok) + encryptedDEK, err := hex.DecodeString(rawDEK) + require.NoError(t, err) + + version, kek := r.keks.Current() + assert.Equal(t, version, client.metadata[config.GetKEKVersionTagName()]) + + plaintext, err := cryptoutil.Decrypt(client.body, encryptedDEK, kek) + require.NoError(t, err) + assert.Equal(t, append(append([]byte{}, part1...), part2...), plaintext) +} + +func TestMultipartCompleteAssemblesInRequestedOrder(t *testing.T) { + r, _ := newBufferRouter(t, "ordering seed") + client := &recordingS3Client{} + + uploadID := createUpload(t, r, client) + uploadPart(t, r, client, uploadPartReq(uploadID, 1, []byte("AAA"))) + uploadPart(t, r, client, uploadPartReq(uploadID, 2, []byte("BBB"))) + + // Complete lists the parts out of upload order; assembly must follow the list. + req := httptest.NewRequest(http.MethodPost, "/bucket/key?uploadId="+uploadID, bytes.NewReader(completeBody(t, 2, 1))) + rec := httptest.NewRecorder() + r.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + require.Equal(t, http.StatusOK, rec.Code) + + encryptedDEK, err := hex.DecodeString(client.metadata[config.GetDekTagName()]) + require.NoError(t, err) + _, kek := r.keks.Current() + plaintext, err := cryptoutil.Decrypt(client.body, encryptedDEK, kek) + require.NoError(t, err) + assert.Equal(t, []byte("BBBAAA"), plaintext) +} + +func TestMultipartUploadPartAWSChunked(t *testing.T) { + r, _ := newBufferRouter(t, "chunked seed") + client := &recordingS3Client{} + + uploadID := createUpload(t, r, client) + + payload := "the quick brown fox" + framed := fmt.Sprintf("%x;chunk-signature=abc\r\n%s\r\n0;chunk-signature=def\r\n\r\n", len(payload), payload) + req := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/bucket/key?partNumber=1&uploadId=%s", uploadID), strings.NewReader(framed)) + req.Header.Set("Content-Encoding", "aws-chunked") + req.Header.Set("x-amz-content-sha256", "STREAMING-AWS4-HMAC-SHA256-PAYLOAD") + uploadPart(t, r, client, req) + + complete := httptest.NewRequest(http.MethodPost, "/bucket/key?uploadId="+uploadID, bytes.NewReader(completeBody(t, 1))) + rec := httptest.NewRecorder() + r.getHandler(complete, client, true, "key", "bucket").ServeHTTP(rec, complete) + require.Equal(t, http.StatusOK, rec.Code) + + encryptedDEK, err := hex.DecodeString(client.metadata[config.GetDekTagName()]) + require.NoError(t, err) + _, kek := r.keks.Current() + plaintext, err := cryptoutil.Decrypt(client.body, encryptedDEK, kek) + require.NoError(t, err) + assert.Equal(t, payload, string(plaintext)) +} + +func TestMultipartAbortCleansUpSession(t *testing.T) { + r, _ := newBufferRouter(t, "abort seed") + client := &recordingS3Client{} + + uploadID := createUpload(t, r, client) + uploadPart(t, r, client, uploadPartReq(uploadID, 1, []byte("data"))) + + abort := httptest.NewRequest(http.MethodDelete, "/bucket/key?uploadId="+uploadID, nil) + rec := httptest.NewRecorder() + r.getHandler(abort, client, true, "key", "bucket").ServeHTTP(rec, abort) + require.Equal(t, http.StatusNoContent, rec.Code) + + // The session is gone: a follow-up UploadPart now reports the upload as unknown. + follow := uploadPartReq(uploadID, 2, []byte("more")) + rec = httptest.NewRecorder() + r.getHandler(follow, client, true, "key", "bucket").ServeHTTP(rec, follow) + assert.Equal(t, http.StatusNotFound, rec.Code) +} + +func TestMultipartCompleteUnknownUploadReturns404(t *testing.T) { + r, _ := newBufferRouter(t, "unknown seed") + client := &recordingS3Client{} + + req := httptest.NewRequest(http.MethodPost, "/bucket/key?uploadId=does-not-exist", bytes.NewReader(completeBody(t, 1))) + rec := httptest.NewRecorder() + r.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + assert.Equal(t, http.StatusNotFound, rec.Code) +} + +func TestMultipartBlockedWithoutManager(t *testing.T) { + r := Router{log: testLogger()} + client := &recordingS3Client{} + + req := httptest.NewRequest(http.MethodPost, "/bucket/key?uploads", nil) + rec := httptest.NewRecorder() + r.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + assert.Equal(t, http.StatusNotImplemented, rec.Code) +} + +func TestAWSChunkedReaderDecodesSingleChunk(t *testing.T) { + payload := "abcdefghij" + framed := fmt.Sprintf("%x;chunk-signature=deadbeef\r\n%s\r\n0;chunk-signature=cafe\r\n\r\n", len(payload), payload) + + out, err := io.ReadAll(newAWSChunkedReader(strings.NewReader(framed))) + require.NoError(t, err) + assert.Equal(t, payload, string(out)) +} + +func TestAWSChunkedReaderDecodesMultipleChunks(t *testing.T) { + framed := "3;chunk-signature=x\r\nabc\r\n2;chunk-signature=y\r\nde\r\n0;chunk-signature=z\r\n\r\n" + + out, err := io.ReadAll(newAWSChunkedReader(strings.NewReader(framed))) + require.NoError(t, err) + assert.Equal(t, "abcde", string(out)) +} diff --git a/s3proxy/internal/router/multipart_xml.go b/s3proxy/internal/router/multipart_xml.go new file mode 100644 index 0000000..3a837d4 --- /dev/null +++ b/s3proxy/internal/router/multipart_xml.go @@ -0,0 +1,153 @@ +/* +Copyright (c) Intrinsec 2026 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package router + +import ( + "bufio" + "encoding/xml" + "fmt" + "io" + "net/http" + "strconv" + "strings" +) + +// InitiateMultipartUploadResult is the XML body returned by CreateMultipartUpload. +// It carries the proxy-generated uploadID the client replays on every subsequent +// UploadPart/Complete/Abort request. +type InitiateMultipartUploadResult struct { + XMLName xml.Name `xml:"InitiateMultipartUploadResult"` + Bucket string `xml:"Bucket"` + Key string `xml:"Key"` + UploadID string `xml:"UploadId"` +} + +// CompleteMultipartUpload is the XML body sent by the client on +// CompleteMultipartUpload. Only the part numbers are used by the proxy: the parts +// are assembled from the on-disk buffer in the listed order. The per-part ETags are +// ignored because the proxy stores the assembled object as a single encrypted +// PutObject and does not reproduce S3's composite ETag. +type CompleteMultipartUpload struct { + XMLName xml.Name `xml:"CompleteMultipartUpload"` + Parts []CompletedPart `xml:"Part"` +} + +// CompletedPart is one entry of a CompleteMultipartUpload part list. +type CompletedPart struct { + PartNumber int `xml:"PartNumber"` + ETag string `xml:"ETag"` +} + +// CompleteMultipartUploadResult is the XML body returned by CompleteMultipartUpload. +// The ETag is the upstream PutObject ETag (of the ciphertext), not S3's +// "md5-of-md5s-N" composite form; standard SDKs do not validate the composite. +type CompleteMultipartUploadResult struct { + XMLName xml.Name `xml:"CompleteMultipartUploadResult"` + Location string `xml:"Location"` + Bucket string `xml:"Bucket"` + Key string `xml:"Key"` + ETag string `xml:"ETag"` +} + +// isAWSChunked reports whether an UploadPart body carries aws-chunked transfer +// framing. SDKs signal this with Content-Encoding: aws-chunked and/or a streaming +// x-amz-content-sha256 marker (STREAMING-AWS4-HMAC-SHA256-PAYLOAD, +// STREAMING-UNSIGNED-PAYLOAD-TRAILER, …). The raw object bytes are recovered by +// stripping the per-chunk size+signature headers. +func isAWSChunked(header http.Header) bool { + if strings.Contains(strings.ToLower(header.Get("Content-Encoding")), "aws-chunked") { + return true + } + return strings.HasPrefix(header.Get("x-amz-content-sha256"), "STREAMING-") +} + +// awsChunkedReader strips aws-chunked transfer framing from an UploadPart body, +// yielding the raw object bytes. The framing is a sequence of +// +// ;chunk-signature=\r\n\r\n +// +// terminated by a zero-size chunk (optionally followed by trailing headers, which +// this reader does not need and leaves unread). Only the hex size before the first +// ';' is significant; the signature is ignored because the upstream request is +// re-signed by the proxy. +type awsChunkedReader struct { + br *bufio.Reader + remaining int64 // unread bytes in the current chunk's data + done bool +} + +func newAWSChunkedReader(r io.Reader) *awsChunkedReader { + return &awsChunkedReader{br: bufio.NewReader(r)} +} + +func (c *awsChunkedReader) Read(p []byte) (int, error) { + for c.remaining == 0 { + if c.done { + return 0, io.EOF + } + if err := c.advance(); err != nil { + return 0, err + } + } + + if int64(len(p)) > c.remaining { + p = p[:c.remaining] + } + n, err := c.br.Read(p) + c.remaining -= int64(n) + if err != nil { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + return n, err + } + if c.remaining == 0 { + if err := c.consumeCRLF(); err != nil { + return n, err + } + } + return n, nil +} + +// advance reads the next chunk header line and sets remaining (or done on the +// terminating zero-size chunk). +func (c *awsChunkedReader) advance() error { + line, err := c.br.ReadString('\n') + if err != nil { + return fmt.Errorf("reading aws-chunked header: %w", err) + } + header := strings.TrimRight(line, "\r\n") + if i := strings.IndexByte(header, ';'); i >= 0 { + header = header[:i] + } + header = strings.TrimSpace(header) + size, err := strconv.ParseInt(header, 16, 64) + if err != nil { + return fmt.Errorf("parsing aws-chunked size %q: %w", header, err) + } + if size < 0 { + return fmt.Errorf("negative aws-chunked size %d", size) + } + if size == 0 { + c.done = true + return nil + } + c.remaining = size + return nil +} + +// consumeCRLF reads and validates the CRLF that terminates a chunk's data. +func (c *awsChunkedReader) consumeCRLF() error { + var crlf [2]byte + if _, err := io.ReadFull(c.br, crlf[:]); err != nil { + return fmt.Errorf("reading aws-chunked CRLF: %w", err) + } + if crlf[0] != '\r' || crlf[1] != '\n' { + return fmt.Errorf("malformed aws-chunked framing: expected CRLF after chunk data") + } + return nil +} diff --git a/s3proxy/internal/router/router.go b/s3proxy/internal/router/router.go index 9815f86..2c7a55e 100644 --- a/s3proxy/internal/router/router.go +++ b/s3proxy/internal/router/router.go @@ -38,6 +38,7 @@ import ( "github.com/intrinsec/s3proxy/internal/config" "github.com/intrinsec/s3proxy/internal/cryptoutil" "github.com/intrinsec/s3proxy/internal/monitoring" + "github.com/intrinsec/s3proxy/internal/multipart" "github.com/intrinsec/s3proxy/internal/s3" ) @@ -53,17 +54,21 @@ type Router struct { keks cryptoutil.KEKProvider client *s3.Client // forwardMultipartReqs controls whether we forward the following requests: CreateMultipartUpload, UploadPart, CompleteMultipartUpload, AbortMultipartUpload. - // s3proxy does not implement those yet. - // Setting forwardMultipartReqs to true will forward those requests to the S3 API, otherwise we block them (secure defaults). + // Setting forwardMultipartReqs to true forwards those requests to the S3 API unchanged (and may leak plaintext upstream). forwardMultipartReqs bool - log *slog.Logger - metrics *monitoring.Metrics + // multipart enables buffer-mode multipart: parts are buffered to local disk and, + // on completion, assembled and stored upstream as a single encrypted PutObject. + // When nil (and not forwarding), multipart requests are blocked (secure default). + // forwardMultipartReqs and multipart are mutually exclusive. + multipart *multipart.Manager + log *slog.Logger + metrics *monitoring.Metrics } // New creates a new Router. The S3 client is built once here and reused for every // incoming request — the AWS SDK client is safe for concurrent use and maintains // its own HTTP connection pool and credentials cache. -func New(ctx context.Context, region string, forwardMultipartReqs bool, log *slog.Logger, metrics *monitoring.Metrics) (Router, error) { +func New(ctx context.Context, region string, forwardMultipartReqs bool, multipartManager *multipart.Manager, log *slog.Logger, metrics *monitoring.Metrics) (Router, error) { seed, err := config.GetEncryptKey() if err != nil { return Router{}, fmt.Errorf("getting encryption key: %w", err) @@ -84,6 +89,7 @@ func New(ctx context.Context, region string, forwardMultipartReqs bool, log *slo keks: keks, client: client, forwardMultipartReqs: forwardMultipartReqs, + multipart: multipartManager, log: log, metrics: metrics, }, nil @@ -119,7 +125,9 @@ func (r Router) Serve(w http.ResponseWriter, req *http.Request) { } // Validate content length for PUT requests: both the absolute S3 ceiling and the - // in-memory PutObject body cap enforced by s3proxy. + // in-memory PutObject body cap enforced by s3proxy. Multipart UploadPart bodies + // are exempt from the 256 MiB PutObject cap (parts are governed by the multipart + // size cap) but still bound by the 5 GiB hard ceiling. if req.Method == http.MethodPut && req.ContentLength > 0 { if err := config.ValidateContentLength(req.ContentLength); err != nil { r.log.Warn("invalid content length", "error", err, "content_length", req.ContentLength) @@ -127,11 +135,13 @@ func (r Router) Serve(w http.ResponseWriter, req *http.Request) { http.Error(w, err.Error(), http.StatusRequestEntityTooLarge) return } - if err := config.ValidatePutBodySize(req.ContentLength); err != nil { - r.log.Warn("put body too large", "error", err, "content_length", req.ContentLength) - r.incError("put_body_too_large") - http.Error(w, err.Error(), http.StatusRequestEntityTooLarge) - return + if !isUploadPart(req.Method, req.URL.Query()) { + if err := config.ValidatePutBodySize(req.ContentLength); err != nil { + r.log.Warn("put body too large", "error", err, "content_length", req.ContentLength) + r.incError("put_body_too_large") + http.Error(w, err.Error(), http.StatusRequestEntityTooLarge) + return + } } } @@ -380,7 +390,7 @@ func (r Router) getHandler(req *http.Request, client s3Client, matchingPath bool } // Check multipart operations first (if not forwarding them) - if handler := r.getMultipartHandler(req); handler != nil { + if handler := r.getMultipartHandler(req, client, key, bucket); handler != nil { return handler } @@ -400,27 +410,40 @@ func (r Router) getHandler(req *http.Request, client s3Client, matchingPath bool return forwardHandler() } -func (r Router) getMultipartHandler(req *http.Request) http.Handler { +// getMultipartHandler selects the handler for a multipart request following the +// precedence forward → buffer → block. In forward mode it returns nil so the +// regular forwarding path handles the request. In buffer mode (a Manager is wired) +// it returns the buffering handler. Otherwise it returns the blocking 501 stub +// (the secure default). ListParts (GET ?uploadId) is not handled here; it stays +// unsupported in v1. +func (r Router) getMultipartHandler(req *http.Request, client s3Client, key, bucket string) http.Handler { if r.forwardMultipartReqs { return nil } - // Check all multipart operations regardless of HTTP method - // Let the is* functions determine if they match + query := req.URL.Query() + buffering := r.multipart != nil - if isUploadPart(req.Method, req.URL.Query()) { - return handleUploadPart(r.log) - } - - if isCreateMultipartUpload(req.Method, req.URL.Query()) { + switch { + case isCreateMultipartUpload(req.Method, query): + if buffering { + return r.handleCreateMultipartUpload(key, bucket) + } return handleCreateMultipartUpload(r.log) - } - - if isCompleteMultipartUpload(req.Method, req.URL.Query()) { + case isUploadPart(req.Method, query): + if buffering { + return r.handleUploadPart(key, bucket) + } + return handleUploadPart(r.log) + case isCompleteMultipartUpload(req.Method, query): + if buffering { + return r.handleCompleteMultipartUpload(client, key, bucket) + } return handleCompleteMultipartUpload(r.log) - } - - if isAbortMultipartUpload(req.Method, req.URL.Query()) { + case isAbortMultipartUpload(req.Method, query): + if buffering { + return r.handleAbortMultipartUpload() + } return handleAbortMultipartUpload(r.log) } From c7fe75a671b3552c6b3fb1737e398b2ed5b9a475 Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Wed, 24 Jun 2026 17:27:43 +0200 Subject: [PATCH 5/9] feat(multipart): activate buffer mode from config + metrics Wire the disk-buffer multipart Manager into main: when S3PROXY_MULTIPART_BUFFER_DIR is set, build the Manager, start its Cleanup goroutine, and pass it to router.New (was nil). Error if --allow-multipart is also set (mutually exclusive); log the single-instance / session-affinity constraint at startup. Add multipart Prometheus collectors (uploads_active, parts_total, buffer_bytes, completed_total, aborted_total, assemble_duration) and make *monitoring.Metrics satisfy the multipart.Metrics interface so the Manager updates them. Increment completed/aborted from the Complete/Abort handlers via nil-safe router helpers. --- s3proxy/cmd/main.go | 37 ++++++++++++- s3proxy/internal/monitoring/monitoring.go | 55 ++++++++++++++++++++ s3proxy/internal/router/handler_multipart.go | 2 + s3proxy/internal/router/router.go | 16 ++++++ 4 files changed, 108 insertions(+), 2 deletions(-) diff --git a/s3proxy/cmd/main.go b/s3proxy/cmd/main.go index 4354d60..69ec70f 100644 --- a/s3proxy/cmd/main.go +++ b/s3proxy/cmd/main.go @@ -13,6 +13,7 @@ package main import ( "context" "crypto/tls" + "errors" "flag" "fmt" "log/slog" @@ -23,6 +24,7 @@ import ( "github.com/intrinsec/s3proxy/internal/config" "github.com/intrinsec/s3proxy/internal/monitoring" + "github.com/intrinsec/s3proxy/internal/multipart" "github.com/intrinsec/s3proxy/internal/router" "github.com/intrinsec/s3proxy/internal/tracing" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" @@ -129,9 +131,14 @@ func runServer(flags cmdFlags, cfg *config.Config, log *slog.Logger) error { metrics := monitoring.New() - // Buffer-mode multipart (the *multipart.Manager) is wired in a later phase; nil + // Buffer-mode multipart is opt-in via S3PROXY_MULTIPART_BUFFER_DIR. A nil manager // keeps multipart blocked by default unless --allow-multipart forwards it. - routerInstance, err := router.New(context.Background(), flags.region, flags.forwardMultipartReqs, nil, log, metrics) + multipartManager, err := newMultipartManager(flags, cfg, log, metrics) + if err != nil { + return err + } + + routerInstance, err := router.New(context.Background(), flags.region, flags.forwardMultipartReqs, multipartManager, log, metrics) if err != nil { return fmt.Errorf("creating router: %w", err) } @@ -199,6 +206,32 @@ func runServer(flags cmdFlags, cfg *config.Config, log *slog.Logger) error { return nil } +// newMultipartManager builds the disk-buffer multipart Manager when +// S3PROXY_MULTIPART_BUFFER_DIR is set, starting its background cleanup goroutine. +// It returns a nil Manager (multipart stays blocked/forwarded) when the dir is +// unset. Buffer mode and --allow-multipart are mutually exclusive. +func newMultipartManager(flags cmdFlags, cfg *config.Config, log *slog.Logger, metrics *monitoring.Metrics) (*multipart.Manager, error) { + dir := cfg.MultipartBufferDir() + if dir == "" { + return nil, nil + } + if flags.forwardMultipartReqs { + return nil, errors.New("S3PROXY_MULTIPART_BUFFER_DIR and --allow-multipart are mutually exclusive") + } + + manager, err := multipart.NewManager(dir, cfg.MultipartMaxSize(), cfg.MultipartTTL(), log, metrics) + if err != nil { + return nil, fmt.Errorf("creating multipart manager: %w", err) + } + + go manager.Cleanup(context.Background()) + + log.Warn("multipart buffer mode enabled: buffers are node-local, so every part of an upload must reach this same instance (single-instance / session-affinity required)", + "buffer_dir", dir, "max_size_bytes", cfg.MultipartMaxSize(), "ttl", cfg.MultipartTTL()) + + return manager, nil +} + func isHealthCheckRequest(r *http.Request) bool { return r.Method == http.MethodGet && (r.URL.Path == "/healthz" || r.URL.Path == "/readyz") } diff --git a/s3proxy/internal/monitoring/monitoring.go b/s3proxy/internal/monitoring/monitoring.go index 29160a6..d0786ac 100644 --- a/s3proxy/internal/monitoring/monitoring.go +++ b/s3proxy/internal/monitoring/monitoring.go @@ -40,6 +40,13 @@ type Metrics struct { DecryptDuration prometheus.Histogram UpstreamErrors prometheus.Counter ThrottledTotal prometheus.Counter + + MultipartUploadsActive prometheus.Gauge + MultipartPartsTotal prometheus.Counter + MultipartBufferBytes prometheus.Gauge + MultipartCompletedTotal prometheus.Counter + MultipartAbortedTotal prometheus.Counter + MultipartAssembleDuration prometheus.Histogram } // New constructs a Metrics bundle backed by a dedicated Registry. @@ -85,6 +92,31 @@ func New() *Metrics { Name: "s3proxy_throttled_total", Help: "Count of requests rejected by the throttling middleware.", }), + MultipartUploadsActive: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "s3proxy_multipart_uploads_active", + Help: "Number of in-flight buffered multipart uploads.", + }), + MultipartPartsTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "s3proxy_multipart_parts_total", + Help: "Count of multipart parts buffered to disk.", + }), + MultipartBufferBytes: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "s3proxy_multipart_buffer_bytes", + Help: "Bytes currently held on disk across all buffered multipart uploads.", + }), + MultipartCompletedTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "s3proxy_multipart_completed_total", + Help: "Count of multipart uploads completed and stored upstream.", + }), + MultipartAbortedTotal: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "s3proxy_multipart_aborted_total", + Help: "Count of multipart uploads aborted by the client.", + }), + MultipartAssembleDuration: prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "s3proxy_multipart_assemble_duration_seconds", + Help: "Time spent assembling buffered multipart parts into a single object.", + Buckets: prometheus.DefBuckets, + }), } reg.MustRegister( @@ -96,11 +128,34 @@ func New() *Metrics { m.DecryptDuration, m.UpstreamErrors, m.ThrottledTotal, + m.MultipartUploadsActive, + m.MultipartPartsTotal, + m.MultipartBufferBytes, + m.MultipartCompletedTotal, + m.MultipartAbortedTotal, + m.MultipartAssembleDuration, ) return m } +// The following methods satisfy the multipart.Metrics interface so the multipart +// Manager can update its buffer gauges and counters without importing this package. + +// AddUploadsActive adjusts the in-flight buffered-upload gauge by delta. +func (m *Metrics) AddUploadsActive(delta float64) { m.MultipartUploadsActive.Add(delta) } + +// IncParts records one buffered multipart part. +func (m *Metrics) IncParts() { m.MultipartPartsTotal.Inc() } + +// AddBufferBytes adjusts the on-disk buffer-bytes gauge by delta. +func (m *Metrics) AddBufferBytes(delta float64) { m.MultipartBufferBytes.Add(delta) } + +// ObserveAssembleSeconds records how long part assembly took. +func (m *Metrics) ObserveAssembleSeconds(seconds float64) { + m.MultipartAssembleDuration.Observe(seconds) +} + // Registry returns the underlying prometheus Registry. Used by tests. func (m *Metrics) Registry() *prometheus.Registry { return m.registry diff --git a/s3proxy/internal/router/handler_multipart.go b/s3proxy/internal/router/handler_multipart.go index ae12f7f..83b36d7 100644 --- a/s3proxy/internal/router/handler_multipart.go +++ b/s3proxy/internal/router/handler_multipart.go @@ -192,6 +192,7 @@ func (r Router) handleCompleteMultipartUpload(client s3Client, key, bucket strin if err := r.multipart.Abort(uploadID); err != nil { log.Warn("CompleteMultipartUpload cleaning up buffer", "upload_id", uploadID, "error", err) } + r.incMultipartCompleted() etag := "" if output.ETag != nil { @@ -218,6 +219,7 @@ func (r Router) handleAbortMultipartUpload() http.HandlerFunc { http.Error(w, "failed to abort multipart upload", http.StatusInternalServerError) return } + r.incMultipartAborted() w.WriteHeader(http.StatusNoContent) } } diff --git a/s3proxy/internal/router/router.go b/s3proxy/internal/router/router.go index 2c7a55e..51733b7 100644 --- a/s3proxy/internal/router/router.go +++ b/s3proxy/internal/router/router.go @@ -506,3 +506,19 @@ func (r Router) incError(class string) { } r.metrics.ErrorsTotal.WithLabelValues(class).Inc() } + +// incMultipartCompleted is a nil-safe helper to bump MultipartCompletedTotal. +func (r Router) incMultipartCompleted() { + if r.metrics == nil { + return + } + r.metrics.MultipartCompletedTotal.Inc() +} + +// incMultipartAborted is a nil-safe helper to bump MultipartAbortedTotal. +func (r Router) incMultipartAborted() { + if r.metrics == nil { + return + } + r.metrics.MultipartAbortedTotal.Inc() +} From 2fb10fc2ea7dbda215547e15db9b795bdbd9be38 Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Wed, 24 Jun 2026 19:08:19 +0200 Subject: [PATCH 6/9] docs(multipart): dashboards, alerts, docs + e2e for buffer mode (phase 6) Operational completeness for buffered multipart uploads. - dashboards: add "Multipart (buffer mode)" row (active uploads, buffer disk bytes, parts/completed/aborted throughput, assemble p95) to both the monitoring/ source and the deployed chart copy - alerts: S3ProxyMultipartBufferHigh (disk exhaustion) and S3ProxyMultipartUploadsStuck (orphaned/affinity-misrouted sessions), tunable via values.yaml - README: new env vars, metrics, alerts, and a Multipart uploads section (forward vs buffer mode, single-instance/affinity, size/memory cap, durability ordering, ETag semantics, ListParts unsupported) - CHANGELOG: Unreleased entry - e2e: Create -> 3x UploadPart -> Complete through the proxy against MinIO; asserts single ciphertext object + DEK tag, byte-equal GET, buffer-dir reclamation - fix: existing proxy_e2e_test.go used the pre-Phase-5 router.New signature (5 args); add the missing nil manager arg so the e2e build compiles --- CHANGELOG.md | 25 +++ README.md | 40 +++- charts/s3proxy/dashboards/s3proxy.json | 78 +++++++ charts/s3proxy/templates/prometheusrule.yaml | 34 +++ charts/s3proxy/values.yaml | 4 + monitoring/alerts/s3proxy.yaml | 33 +++ monitoring/dashboards/s3proxy.json | 72 +++++++ s3proxy/internal/e2e/multipart_e2e_test.go | 210 +++++++++++++++++++ s3proxy/internal/e2e/proxy_e2e_test.go | 3 +- 9 files changed, 495 insertions(+), 4 deletions(-) create mode 100644 s3proxy/internal/e2e/multipart_e2e_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index e779d8f..4aec153 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,31 @@ Two version streams move independently: ## [Unreleased] +### Added + +- **Buffered multipart uploads** (opt-in via `S3PROXY_MULTIPART_BUFFER_DIR`). + s3proxy now accepts the full multipart protocol, buffers each part to a + node-local disk directory, and on `CompleteMultipartUpload` concatenates the + parts, encrypts the whole object through the existing AES-256-GCM envelope + path, and stores it upstream as a **single ciphertext PutObject** — so data at + rest stays encrypted (unlike `--allow-multipart`, which forwards plaintext). + - New `internal/multipart` disk-buffer session manager with a background + sweeper that evicts idle uploads past `S3PROXY_MULTIPART_TTL` (default 24h) + and reclaims orphaned directories left by a restart. + - New config knobs: `S3PROXY_MULTIPART_BUFFER_DIR` (enables the mode), + `S3PROXY_MULTIPART_MAX_SIZE` (assembled-object cap, default/hard cap 5 GiB), + `S3PROXY_MULTIPART_TTL`. Mutually exclusive with `--allow-multipart`. + - New metrics: `s3proxy_multipart_uploads_active`, + `s3proxy_multipart_buffer_bytes`, `s3proxy_multipart_parts_total`, + `s3proxy_multipart_completed_total`, `s3proxy_multipart_aborted_total`, + `s3proxy_multipart_assemble_duration_seconds`; Grafana dashboard row and two + `PrometheusRule` alerts (`S3ProxyMultipartBufferHigh`, + `S3ProxyMultipartUploadsStuck`). + - **Constraints (documented):** single-instance / session-affinity required + (buffers are node-local), assembled object must fit in RAM (~2× peak), + `ListParts` unsupported. `Complete` acks only after the upstream store + succeeds, inheriting the single-shot path's durability ordering. + ### Chart - **`chart/1.9.3`** — Dashboard usability + Go runtime panels. diff --git a/README.md b/README.md index 8efcbe4..88e0da7 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ The project is a hardened fork of the original [edgelesssys/constellation](https - [Logs](#logs) - [Health probes](#health-probes) - [Troubleshooting](#troubleshooting) +- [Multipart uploads](#multipart-uploads) - [Security](#security) - [Architecture](#architecture) - [Helm chart reference](#helm-chart-reference) @@ -89,6 +90,9 @@ All runtime configuration is sourced from environment variables. The Helm chart | `AWS_SECRET_ACCESS_KEY` | yes | — | Idem. | | `S3PROXY_THROTTLING_REQUESTSMAX` | no | `0` (off) | Cap on **concurrent in-flight requests** (not RPS). Excess requests are rejected. | | `S3PROXY_PUTBODY_MAX` | no | `268435456` (256 MiB) | Per-request PutObject body size ceiling, in bytes. Up to `5368709120` (5 GiB, the S3 hard cap). | +| `S3PROXY_MULTIPART_BUFFER_DIR` | no | unset (disabled) | Enables **buffer-mode multipart**: parts are buffered to this local directory and assembled into a single encrypted PutObject on completion. Empty leaves multipart blocked. Mutually exclusive with `--allow-multipart`. **Single-instance / session-affinity only** — see [Multipart uploads](#multipart-uploads). | +| `S3PROXY_MULTIPART_MAX_SIZE` | no | `5368709120` (5 GiB) | Maximum assembled object size for a buffered multipart upload, in bytes. Clamped to 5 GiB (the S3 hard cap). Bounds peak RAM (~2× this) at completion. | +| `S3PROXY_MULTIPART_TTL` | no | `24h` | How long an idle buffered upload (and its on-disk parts) is retained before the background sweeper evicts it. Also bounds how long orphaned directories survive a restart. | | `S3PROXY_DEKTAG_NAME` | no | `isec` | S3 object-metadata key used to store the encrypted DEK. | | `S3PROXY_DEKTAG_KEKVER` | no | `-kek-ver` | S3 object-metadata key recording which KEK derivation version wrapped the DEK. | | `S3PROXY_INSECURE` | no | unset | Set to `1` to use plain HTTP (not HTTPS) when talking to upstream. **Dev / e2e only.** Emits a loud warning at startup. | @@ -128,6 +132,12 @@ S3Proxy exposes Prometheus metrics on `/metrics` (no authentication; scope via N | `s3proxy_decrypt_duration_seconds` | histogram | — | Time spent decrypting GetObject bodies. | | `s3proxy_upstream_errors_total` | counter | — | Errors talking to the upstream S3 endpoint. | | `s3proxy_throttled_total` | counter | — | Requests rejected by the throttling middleware. | +| `s3proxy_multipart_uploads_active` | gauge | — | In-flight buffered multipart uploads (buffer mode only). | +| `s3proxy_multipart_buffer_bytes` | gauge | — | Bytes currently held on disk across all buffered multipart uploads. | +| `s3proxy_multipart_parts_total` | counter | — | Multipart parts buffered to disk. | +| `s3proxy_multipart_completed_total` | counter | — | Multipart uploads completed and stored upstream. | +| `s3proxy_multipart_aborted_total` | counter | — | Multipart uploads aborted by the client. | +| `s3proxy_multipart_assemble_duration_seconds` | histogram | — | Time spent assembling buffered parts into one object at completion. | The default Go and process collectors (`go_*`, `process_*`) are also registered. @@ -142,7 +152,7 @@ serviceMonitor: ### Alerts -The chart ships a `PrometheusRule` (off by default) with four alerts. All thresholds and `for` windows are tunable via `values.yaml`. +The chart ships a `PrometheusRule` (off by default) with six alerts. All thresholds and `for` windows are tunable via `values.yaml`. | Alert | Severity | PromQL (default threshold) | Default `for` | |---|---|---|---| @@ -150,8 +160,10 @@ The chart ships a `PrometheusRule` (off by default) with four alerts. All thresh | `S3ProxyHighLatency` | warning | `histogram_quantile(0.95, sum by (le) (rate(http_request_duration_seconds_bucket{service="s3proxy"}[5m]))) > 2` | `10m` | | `S3ProxyServiceDown` | critical | `up{job=~".*s3proxy.*"} == 0` | `2m` | | `S3ProxyHighCrashRate` | critical | `increase(service_crashes_total{service="s3proxy"}[5m]) > 0` | `5m` | +| `S3ProxyMultipartBufferHigh` | warning | `s3proxy_multipart_buffer_bytes{service="s3proxy"} > 10737418240` (10 GiB) | `15m` | +| `S3ProxyMultipartUploadsStuck` | warning | `min_over_time(s3proxy_multipart_uploads_active{service="s3proxy"}[2h]) > 0` | `15m` | -Enable with: +The two multipart alerts only fire when buffer-mode multipart is enabled (`S3PROXY_MULTIPART_BUFFER_DIR`); otherwise the metrics stay at zero. Enable with: ```yaml prometheusRule: @@ -162,6 +174,8 @@ prometheusRule: highErrorRate: 0.05 highLatencySeconds: 2 crashes: 0 + multipartBufferBytes: 10737418240 + multipartStuckUploads: 0 ``` ### Tracing @@ -203,6 +217,26 @@ args: ["--level=-1"] # Debug, use only for troubleshooting --- +## Multipart uploads + +Multipart uploads are **blocked by default** (the four multipart endpoints return `501`). Many SDKs auto-switch to multipart above a size threshold, so large uploads fail unless one of two opt-in modes is enabled: + +- **Forward mode** (`--allow-multipart`) — parts are forwarded to the upstream **unencrypted**. Leaks plaintext at rest; avoid unless you accept that. +- **Buffer mode** (`S3PROXY_MULTIPART_BUFFER_DIR`) — parts are buffered to local disk and, on `CompleteMultipartUpload`, concatenated, encrypted (the same AES-256-GCM envelope path as single PutObject) and written upstream as a **single ciphertext object**. Data at rest stays encrypted. + +The two modes are mutually exclusive; setting both is a startup error. + +### Buffer-mode constraints + +- **Single instance / session affinity (hard requirement).** Buffers and the in-flight-upload table are node-local. Every `UploadPart` and the final `CompleteMultipartUpload` for an upload **must reach the same pod that created it**. Run a single replica, or pin client sessions to one pod (e.g. sticky sessions / `sessionAffinity: ClientIP`). There is no distributed/shared state. +- **Size and memory cap.** The crypto path is two-pass and cannot stream, so the whole object is assembled in RAM at completion (peak ≈ 2× object size). `S3PROXY_MULTIPART_MAX_SIZE` (default and hard cap 5 GiB) bounds this; oversized parts/objects are rejected with `413`. Size the pod's memory accordingly. +- **Disk usage.** Buffered parts occupy `S3PROXY_MULTIPART_BUFFER_DIR` until the upload completes or its TTL expires. A background sweeper evicts uploads idle longer than `S3PROXY_MULTIPART_TTL` (default 24h) and, on restart, reclaims orphaned directories left by the lost in-memory table. Watch `s3proxy_multipart_buffer_bytes` and the `S3ProxyMultipartBufferHigh` alert; provision the volume for your concurrency × object size. +- **Durability / ack ordering.** `CompleteMultipartUpload` returns `200` only after the upstream `PutObject` succeeds (using a detached context so a client disconnect cannot abort an in-flight store). An `UploadPart` `200` means "part buffered", not "object stored" — identical to real S3. A crash before upstream success fails `Complete` (the client retries); a crash after success but before the `200` leaves the object durably stored, so a retry is at worst a redundant overwrite. +- **ETags.** Per-part ETags are synthesized MD5s and the final object ETag is the upstream object's ETag, **not** S3's composite `md5-of-md5s-N` form. Standard SDKs do not validate the composite, so this is transparent in practice. +- **`ListParts` is not supported** in buffer mode (returns `501`); SDK upload managers that complete from their own tracked part list are unaffected. + +--- + ## Security S3Proxy is designed for at-rest data protection in front of an S3-compatible backend that you do not fully trust (multi-tenant object storage, third-party cloud, …). It is **not** a substitute for IAM, network controls, or end-to-end client-side encryption. @@ -211,7 +245,7 @@ S3Proxy is designed for at-rest data protection in front of an S3-compatible bac - **Key wrapping** — per-object DEKs are wrapped with NIST SP-800-38F Key-Wrap-with-Padding (KWP). - **KEK derivation** — the seed (`S3PROXY_ENCRYPT_KEY`) is hardened with HKDF-SHA256; the derivation version is recorded on every object (`-kek-ver` metadata) so future KEK rotations stay backward-compatible. - **Body cap** — PutObject bodies are bounded to `S3PROXY_PUTBODY_MAX` (default 256 MiB, hard cap 5 GiB) to limit memory pressure and DoS surface. -- **Multipart blocked by default** — the four multipart endpoints (`CreateMultipartUpload`, `UploadPart`, `CompleteMultipartUpload`, `AbortMultipartUpload`) are refused unless `--allow-multipart` is passed, because part-by-part data cannot be encrypted in the current design. **Enabling `--allow-multipart` will store unencrypted data on the upstream** — only use it when you understand and accept that. +- **Multipart blocked by default** — the four multipart endpoints (`CreateMultipartUpload`, `UploadPart`, `CompleteMultipartUpload`, `AbortMultipartUpload`) are refused unless either multipart mode is opted into. **`--allow-multipart` forwards parts to the upstream unencrypted** — only use it when you understand and accept that. For encrypted multipart, enable **buffer mode** (`S3PROXY_MULTIPART_BUFFER_DIR`) instead: parts are buffered locally and stored upstream as a single ciphertext PutObject, so data at rest stays encrypted. See [Multipart uploads](#multipart-uploads) for its constraints. The two modes are mutually exclusive. - **Upstream HTTPS by default** — `S3PROXY_INSECURE=1` is a dev/e2e knob and emits a loud warning at startup. Production deployments must leave it unset. - **Decryption fallback** — `S3PROXY_DECRYPTION_FALLBACK=1` re-tries decryption with an all-zero KEK to bridge migrations away from unencrypted data. It is a migration helper, not a steady-state option; switch it back off once the migration finishes. - **KEK rotation** — currently a single active KEK is supported. The KEK-version metadata is in place for a future multi-KEK rotation flow, but operational KEK rotation is not yet implemented. diff --git a/charts/s3proxy/dashboards/s3proxy.json b/charts/s3proxy/dashboards/s3proxy.json index 06405eb..f79a0ee 100644 --- a/charts/s3proxy/dashboards/s3proxy.json +++ b/charts/s3proxy/dashboards/s3proxy.json @@ -273,6 +273,84 @@ } ], "fieldConfig": {"defaults": {"unit": "short", "min": 0}} + }, + { + "id": 14, + "type": "row", + "title": "Multipart (buffer mode)", + "gridPos": {"x": 0, "y": 49, "w": 24, "h": 1} + }, + { + "id": 15, + "type": "timeseries", + "title": "Active buffered uploads", + "gridPos": {"x": 0, "y": 50, "w": 12, "h": 8}, + "targets": [ + { + "datasource": {"type": "prometheus", "uid": "${datasource}"}, + "expr": "s3proxy_multipart_uploads_active{job=~\"$job\", instance=~\"$instance\"}", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "fieldConfig": {"defaults": {"unit": "short", "min": 0}} + }, + { + "id": 16, + "type": "timeseries", + "title": "Buffer disk usage", + "gridPos": {"x": 12, "y": 50, "w": 12, "h": 8}, + "targets": [ + { + "datasource": {"type": "prometheus", "uid": "${datasource}"}, + "expr": "s3proxy_multipart_buffer_bytes{job=~\"$job\", instance=~\"$instance\"}", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "fieldConfig": {"defaults": {"unit": "bytes", "min": 0}} + }, + { + "id": 17, + "type": "timeseries", + "title": "Multipart throughput (parts / completed / aborted)", + "gridPos": {"x": 0, "y": 58, "w": 12, "h": 8}, + "targets": [ + { + "datasource": {"type": "prometheus", "uid": "${datasource}"}, + "expr": "sum(rate(s3proxy_multipart_parts_total{job=~\"$job\", instance=~\"$instance\"}[5m]))", + "legendFormat": "parts/s", + "refId": "A" + }, + { + "datasource": {"type": "prometheus", "uid": "${datasource}"}, + "expr": "sum(rate(s3proxy_multipart_completed_total{job=~\"$job\", instance=~\"$instance\"}[5m]))", + "legendFormat": "completed/s", + "refId": "B" + }, + { + "datasource": {"type": "prometheus", "uid": "${datasource}"}, + "expr": "sum(rate(s3proxy_multipart_aborted_total{job=~\"$job\", instance=~\"$instance\"}[5m]))", + "legendFormat": "aborted/s", + "refId": "C" + } + ], + "fieldConfig": {"defaults": {"unit": "ops"}} + }, + { + "id": 18, + "type": "timeseries", + "title": "Assemble duration (p95)", + "gridPos": {"x": 12, "y": 58, "w": 12, "h": 8}, + "targets": [ + { + "datasource": {"type": "prometheus", "uid": "${datasource}"}, + "expr": "histogram_quantile(0.95, sum by (le) (rate(s3proxy_multipart_assemble_duration_seconds_bucket{job=~\"$job\", instance=~\"$instance\"}[5m])))", + "legendFormat": "assemble p95", + "refId": "A" + } + ], + "fieldConfig": {"defaults": {"unit": "s"}} } ] } diff --git a/charts/s3proxy/templates/prometheusrule.yaml b/charts/s3proxy/templates/prometheusrule.yaml index 12a8598..df14c15 100644 --- a/charts/s3proxy/templates/prometheusrule.yaml +++ b/charts/s3proxy/templates/prometheusrule.yaml @@ -74,4 +74,38 @@ spec: service_crashes_total has increased in the last 5 minutes, meaning the HTTP middleware recovered from at least one panic. Inspect recent logs for stack traces. + + - alert: S3ProxyMultipartBufferHigh + expr: | + s3proxy_multipart_buffer_bytes{service="s3proxy"} + > {{ .Values.prometheusRule.thresholds.multipartBufferBytes | default 10737418240 }} + for: {{ .Values.prometheusRule.for.multipartBufferHigh | default "15m" }} + labels: + severity: warning + service: s3proxy + annotations: + summary: "s3proxy multipart buffer above threshold" + description: | + On-disk multipart buffer usage has exceeded the configured + threshold. Buffered parts live on node-local disk until their + upload completes or its TTL expires; sustained growth risks + disk exhaustion. Check for stuck uploads and buffer-dir space. + + - alert: S3ProxyMultipartUploadsStuck + expr: | + min_over_time(s3proxy_multipart_uploads_active{service="s3proxy"}[2h]) + > {{ .Values.prometheusRule.thresholds.multipartStuckUploads | default 0 }} + for: {{ .Values.prometheusRule.for.multipartUploadsStuck | default "15m" }} + labels: + severity: warning + service: s3proxy + annotations: + summary: "s3proxy has long-lived buffered multipart uploads" + description: | + One or more buffered multipart uploads have stayed in-flight for + over 2 hours without completing or aborting. Buffers are + node-local, so this usually means orphaned sessions (a client + that never completed, or parts routed to the wrong instance + without session affinity). They are reclaimed at the configured + TTL; investigate before disk fills. {{- end }} diff --git a/charts/s3proxy/values.yaml b/charts/s3proxy/values.yaml index 4cc25b1..3553bdc 100644 --- a/charts/s3proxy/values.yaml +++ b/charts/s3proxy/values.yaml @@ -162,11 +162,15 @@ prometheusRule: highErrorRate: 0.05 # fraction of 5xx responses → alert when >5% highLatencySeconds: 2 # p99 latency seconds → alert when >2s crashes: 0 # container restart count → alert when >0 + multipartBufferBytes: 10737418240 # multipart buffer disk bytes → alert when >10 GiB + multipartStuckUploads: 0 # uploads buffered >2h continuously → alert when >0 for: highErrorRate: 10m # sustained window before firing highLatency: 10m serviceDown: 2m highCrashRate: 5m + multipartBufferHigh: 15m + multipartUploadsStuck: 15m # Grafana dashboard CRD. When enabled the chart ships a `GrafanaDashboard` # (`grafana.integreatly.org/v1beta1`) carrying the s3proxy dashboard JSON. diff --git a/monitoring/alerts/s3proxy.yaml b/monitoring/alerts/s3proxy.yaml index af84be9..a35ccd7 100644 --- a/monitoring/alerts/s3proxy.yaml +++ b/monitoring/alerts/s3proxy.yaml @@ -59,3 +59,36 @@ groups: service_crashes_total has increased in the last 5 minutes, meaning the HTTP middleware recovered from at least one panic. Inspect recent logs for stack traces and root-cause the failure. + + - alert: S3ProxyMultipartBufferHigh + expr: | + s3proxy_multipart_buffer_bytes{service="s3proxy"} > 10737418240 + for: 15m + labels: + severity: warning + service: s3proxy + annotations: + summary: "s3proxy multipart buffer above 10 GiB" + description: | + On-disk multipart buffer usage has stayed above 10 GiB for 15 + minutes. Buffered parts live on node-local disk until their upload + completes or its TTL expires; sustained growth risks disk + exhaustion. Check for stuck/abandoned uploads and the buffer-dir + free space. + + - alert: S3ProxyMultipartUploadsStuck + expr: | + min_over_time(s3proxy_multipart_uploads_active{service="s3proxy"}[2h]) > 0 + for: 15m + labels: + severity: warning + service: s3proxy + annotations: + summary: "s3proxy has multipart uploads buffered for over 2h" + description: | + At least one buffered multipart upload has stayed in-flight for + more than 2 hours without completing or aborting. Buffers are + node-local, so this usually means orphaned sessions (a client that + never sent CompleteMultipartUpload, or parts routed to the wrong + instance without session affinity). They are reclaimed at the + configured TTL; investigate before disk fills. diff --git a/monitoring/dashboards/s3proxy.json b/monitoring/dashboards/s3proxy.json index 972335e..23c85c9 100644 --- a/monitoring/dashboards/s3proxy.json +++ b/monitoring/dashboards/s3proxy.json @@ -126,6 +126,78 @@ } ], "fieldConfig": {"defaults": {"unit": "ops"}} + }, + { + "id": 7, + "type": "row", + "title": "Multipart (buffer mode)", + "gridPos": {"x": 0, "y": 24, "w": 24, "h": 1} + }, + { + "id": 8, + "type": "timeseries", + "title": "Active buffered uploads", + "gridPos": {"x": 0, "y": 25, "w": 12, "h": 8}, + "targets": [ + { + "expr": "s3proxy_multipart_uploads_active{instance=~\"$instance\"}", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "fieldConfig": {"defaults": {"unit": "short", "min": 0}} + }, + { + "id": 9, + "type": "timeseries", + "title": "Buffer disk usage", + "gridPos": {"x": 12, "y": 25, "w": 12, "h": 8}, + "targets": [ + { + "expr": "s3proxy_multipart_buffer_bytes{instance=~\"$instance\"}", + "legendFormat": "{{instance}}", + "refId": "A" + } + ], + "fieldConfig": {"defaults": {"unit": "bytes", "min": 0}} + }, + { + "id": 10, + "type": "timeseries", + "title": "Multipart throughput (parts / completed / aborted)", + "gridPos": {"x": 0, "y": 33, "w": 12, "h": 8}, + "targets": [ + { + "expr": "rate(s3proxy_multipart_parts_total{instance=~\"$instance\"}[5m])", + "legendFormat": "parts/s", + "refId": "A" + }, + { + "expr": "rate(s3proxy_multipart_completed_total{instance=~\"$instance\"}[5m])", + "legendFormat": "completed/s", + "refId": "B" + }, + { + "expr": "rate(s3proxy_multipart_aborted_total{instance=~\"$instance\"}[5m])", + "legendFormat": "aborted/s", + "refId": "C" + } + ], + "fieldConfig": {"defaults": {"unit": "ops"}} + }, + { + "id": 11, + "type": "timeseries", + "title": "Assemble duration (p95)", + "gridPos": {"x": 12, "y": 33, "w": 12, "h": 8}, + "targets": [ + { + "expr": "histogram_quantile(0.95, sum by (le) (rate(s3proxy_multipart_assemble_duration_seconds_bucket{instance=~\"$instance\"}[5m])))", + "legendFormat": "assemble p95", + "refId": "A" + } + ], + "fieldConfig": {"defaults": {"unit": "s"}} } ] } diff --git a/s3proxy/internal/e2e/multipart_e2e_test.go b/s3proxy/internal/e2e/multipart_e2e_test.go new file mode 100644 index 0000000..2a22e8b --- /dev/null +++ b/s3proxy/internal/e2e/multipart_e2e_test.go @@ -0,0 +1,210 @@ +//go:build e2e + +/* +Copyright (c) Intrinsec 2026 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package e2e + +import ( + "bytes" + "context" + "crypto/rand" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/intrinsec/s3proxy/internal/config" + "github.com/intrinsec/s3proxy/internal/monitoring" + "github.com/intrinsec/s3proxy/internal/multipart" + "github.com/intrinsec/s3proxy/internal/router" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + tcminio "github.com/testcontainers/testcontainers-go/modules/minio" +) + +// TestMultipartBufferModeRoundtrip spins up a real MinIO container with buffer-mode +// multipart enabled and drives a full Create → UploadPart × N → Complete sequence +// through the proxy. It asserts that: +// +// - the parts are buffered to disk and assembled into a *single* upstream object +// (MinIO holds exactly one object, never per-part objects); +// - that object is ciphertext at rest and carries the DEK metadata tag; +// - a GetObject through the proxy returns the original plaintext byte-for-byte; +// - the on-disk buffer directory is reclaimed once the upload completes. +// +// The SDK frames UploadPart bodies as aws-chunked by default, so this also exercises +// the proxy's chunk de-framer on the real client wire format. +func TestMultipartBufferModeRoundtrip(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + + minioCtr, err := tcminio.Run(ctx, "minio/minio:RELEASE.2025-09-07T16-13-09Z") + if err != nil { + t.Skipf("cannot start minio testcontainer (docker unavailable?): %v", err) + } + t.Cleanup(func() { + if err := minioCtr.Terminate(context.Background()); err != nil { + t.Logf("terminate minio container: %v", err) + } + }) + + endpoint, err := minioCtr.ConnectionString(ctx) + require.NoError(t, err) + + bufferDir := t.TempDir() + + t.Setenv("S3PROXY_HOST", endpoint) + t.Setenv("S3PROXY_INSECURE", "1") + t.Setenv("S3PROXY_ENCRYPT_KEY", "e2e-multipart-seed") + t.Setenv("S3PROXY_MULTIPART_BUFFER_DIR", bufferDir) + t.Setenv("AWS_ACCESS_KEY_ID", minioCtr.Username) + t.Setenv("AWS_SECRET_ACCESS_KEY", minioCtr.Password) + t.Setenv("AWS_REGION", "us-east-1") + + require.NoError(t, config.LoadConfig()) + require.NotEmpty(t, config.GetMultipartBufferDir(), "buffer mode must be enabled for this test") + + bucket := "s3proxy-multipart-e2e" + key := "large/payload.bin" + + // 12 MiB of incompressible data, split into three parts. Random bytes guarantee + // the at-rest object cannot accidentally equal the plaintext. + plaintext := make([]byte, 12<<20) + _, err = rand.Read(plaintext) + require.NoError(t, err) + const partSize = 4 << 20 + + directCfg, err := awsconfig.LoadDefaultConfig(ctx, + awsconfig.WithRegion("us-east-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(minioCtr.Username, minioCtr.Password, "")), + ) + require.NoError(t, err) + + // Direct MinIO client: creates the bucket and inspects what lands at rest. + directClient := awss3.NewFromConfig(directCfg, func(o *awss3.Options) { + o.UsePathStyle = true + o.BaseEndpoint = aws.String("http://" + endpoint) + o.ResponseChecksumValidation = aws.ResponseChecksumValidationUnset + }) + _, err = directClient.CreateBucket(ctx, &awss3.CreateBucketInput{Bucket: aws.String(bucket)}) + require.NoError(t, err) + + // Proxy wiring with a live multipart Manager backed by the temp buffer dir. + log := slog.New(slog.NewJSONHandler(io.Discard, nil)) + metrics := monitoring.New() + + manager, err := multipart.NewManager(bufferDir, config.GetMultipartMaxSize(), config.GetMultipartTTL(), log, metrics) + require.NoError(t, err) + + routerInstance, err := router.New(ctx, "us-east-1", false, manager, log, metrics) + require.NoError(t, err) + + mux := http.NewServeMux() + mux.Handle(monitoring.MetricsPath, metrics.Handler()) + mux.HandleFunc("/", routerInstance.Serve) + proxySrv := httptest.NewServer(metrics.Instrument(mux)) + t.Cleanup(proxySrv.Close) + + proxyClient := awss3.NewFromConfig(directCfg, func(o *awss3.Options) { + o.UsePathStyle = true + o.BaseEndpoint = aws.String(proxySrv.URL) + // The proxy returns decrypted plaintext and cannot forward an at-rest + // checksum, so disable response validation on the test client. + o.ResponseChecksumValidation = aws.ResponseChecksumValidationUnset + }) + + // Create → UploadPart × 3 → Complete, all through the proxy. + created, err := proxyClient.CreateMultipartUpload(ctx, &awss3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + ContentType: aws.String("application/octet-stream"), + }) + require.NoError(t, err) + require.NotNil(t, created.UploadId) + uploadID := *created.UploadId + + var completedParts []s3types.CompletedPart + for partNumber, off := int32(1), 0; off < len(plaintext); partNumber++ { + end := min(off+partSize, len(plaintext)) + out, err := proxyClient.UploadPart(ctx, &awss3.UploadPartInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: aws.String(uploadID), + PartNumber: aws.Int32(partNumber), + Body: bytes.NewReader(plaintext[off:end]), + }) + require.NoError(t, err) + require.NotNil(t, out.ETag) + completedParts = append(completedParts, s3types.CompletedPart{ + ETag: out.ETag, + PartNumber: aws.Int32(partNumber), + }) + off = end + } + require.Len(t, completedParts, 3, "12 MiB at 4 MiB parts must produce three parts") + + _, err = proxyClient.CompleteMultipartUpload(ctx, &awss3.CompleteMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: aws.String(uploadID), + MultipartUpload: &s3types.CompletedMultipartUpload{Parts: completedParts}, + }) + require.NoError(t, err) + + // MinIO must hold exactly one object: the single assembled ciphertext, never + // per-part objects (the parts never reach the backend). + listed, err := directClient.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{Bucket: aws.String(bucket)}) + require.NoError(t, err) + require.Len(t, listed.Contents, 1, "buffered multipart must store exactly one upstream object") + require.NotNil(t, listed.Contents[0].Key) + assert.Equal(t, key, *listed.Contents[0].Key) + + // That object is ciphertext at rest and carries the DEK metadata tag. + stored, err := directClient.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + require.NoError(t, err) + storedBody, err := io.ReadAll(stored.Body) + require.NoError(t, err) + require.NoError(t, stored.Body.Close()) + + assert.NotEqual(t, plaintext, storedBody, "stored bytes must differ from plaintext") + assert.False(t, bytes.Contains(storedBody, plaintext), "plaintext must not appear verbatim at rest") + assert.Greater(t, len(storedBody), len(plaintext), "ciphertext must be longer than plaintext (nonce + auth tag)") + assert.NotEmpty(t, stored.Metadata[config.GetDekTagName()], "DEK metadata tag must be attached") + assert.NotEmpty(t, stored.Metadata[config.GetKEKVersionTagName()], "KEK version metadata tag must be attached") + + // Round-trip through the proxy returns the original plaintext. + got, err := proxyClient.GetObject(ctx, &awss3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + require.NoError(t, err) + gotBody, err := io.ReadAll(got.Body) + require.NoError(t, err) + require.NoError(t, got.Body.Close()) + assert.Equal(t, plaintext, gotBody, "GetObject through the proxy must return the original plaintext") + + // The on-disk buffer is reclaimed once the upload completes (the session dir is + // removed; only the base buffer dir remains). + entries, err := os.ReadDir(bufferDir) + require.NoError(t, err) + assert.Empty(t, entries, "buffer directory must be empty after CompleteMultipartUpload") + + // The completion counter is exposed on /metrics. + metricsResp, err := http.Get(proxySrv.URL + monitoring.MetricsPath) + require.NoError(t, err) + metricsBody, err := io.ReadAll(metricsResp.Body) + require.NoError(t, err) + require.NoError(t, metricsResp.Body.Close()) + assert.True(t, strings.Contains(string(metricsBody), "s3proxy_multipart_completed_total"), + "metrics endpoint must expose s3proxy_multipart_completed_total") +} diff --git a/s3proxy/internal/e2e/proxy_e2e_test.go b/s3proxy/internal/e2e/proxy_e2e_test.go index f3fa1db..3c78a85 100644 --- a/s3proxy/internal/e2e/proxy_e2e_test.go +++ b/s3proxy/internal/e2e/proxy_e2e_test.go @@ -101,7 +101,8 @@ func TestProxyRoundtripAgainstMinio(t *testing.T) { log := slog.New(slog.NewJSONHandler(io.Discard, nil)) metrics := monitoring.New() - routerInstance, err := router.New(ctx, "us-east-1", false, log, metrics) + // nil multipart manager: this test exercises only the single-shot PutObject path. + routerInstance, err := router.New(ctx, "us-east-1", false, nil, log, metrics) require.NoError(t, err) mux := http.NewServeMux() From 1bf2a1ce3a2ad9b95a9876b03521f2803d6edcac Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Thu, 25 Jun 2026 09:14:41 +0200 Subject: [PATCH 7/9] fix(router): emit Content-Length on intercepted GetObject responses Intercepted GetObject responses streamed the decrypted body without a Content-Length header, so Go's server fell back to chunked transfer encoding. s3cmd 2.4.0 reads content-length unconditionally and downloaded an empty file as a result. The full plaintext is already buffered before any bytes are written, so set Content-Length from len(plaintext) before WriteHeader. The upstream length describes the ciphertext (+28 bytes GCM overhead, or arbitrary for legacy objects) and can't be reused, mirroring the existing omission of x-amz-checksum-* headers. --- CHANGELOG.md | 7 ++++++ s3proxy/internal/router/object.go | 6 ++++++ s3proxy/internal/router/router_test.go | 30 ++++++++++++++++++++++++++ 3 files changed, 43 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4aec153..53d80de 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,13 @@ Two version streams move independently: `ListParts` unsupported. `Complete` acks only after the upstream store succeeds, inheriting the single-shot path's durability ordering. +### Fixed +- **s3cmd `get` compatibility**: intercepted `GetObject` responses now + carry a `Content-Length` header reflecting the decrypted body size. + Previously the proxy streamed decrypted objects without a length, so + Go fell back to chunked transfer encoding and s3cmd 2.4.0 downloaded + an empty file. Downloads now complete with the correct size. + ### Chart - **`chart/1.9.3`** — Dashboard usability + Go runtime panels. diff --git a/s3proxy/internal/router/object.go b/s3proxy/internal/router/object.go index abb1a24..f5e4cb0 100644 --- a/s3proxy/internal/router/object.go +++ b/s3proxy/internal/router/object.go @@ -16,6 +16,7 @@ import ( "net/http" "net/url" "runtime/debug" + "strconv" "strings" "syscall" "time" @@ -157,6 +158,11 @@ func (o object) get(w http.ResponseWriter, r *http.Request) { releaseLargeBuffer(&plaintext) return default: + // The full plaintext is already buffered, so emit an explicit Content-Length + // (the upstream value describes the ciphertext, not the decrypted body) so the + // server uses identity encoding instead of chunked. Some clients (e.g. s3cmd) + // require a Content-Length header on the response. + w.Header().Set("Content-Length", strconv.Itoa(len(plaintext))) w.WriteHeader(http.StatusOK) _, writeErr := w.Write(plaintext) releaseLargeBuffer(&plaintext) diff --git a/s3proxy/internal/router/router_test.go b/s3proxy/internal/router/router_test.go index 2eae233..a445ceb 100644 --- a/s3proxy/internal/router/router_test.go +++ b/s3proxy/internal/router/router_test.go @@ -13,6 +13,7 @@ import ( "log/slog" "net/http" "net/http/httptest" + "strconv" "strings" "testing" "time" @@ -198,6 +199,35 @@ func TestGetObjectUsesRouterKEK(t *testing.T) { assert.Equal(t, "secret payload", rec.Body.String()) } +func TestGetObjectSetsPlaintextContentLength(t *testing.T) { + keks := newTestKEKs(t, "expected encryption key") + version, curKEK := keks.Current() + client := newEncryptedGetObjectClient(t, curKEK, version, []byte("secret payload")) + router := Router{keks: keks, log: testLogger()} + req := httptest.NewRequest(http.MethodGet, "/bucket/key", nil) + rec := httptest.NewRecorder() + + router.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + + require.Equal(t, http.StatusOK, rec.Code) + // The header must reflect the plaintext length, not the +28-byte ciphertext. + assert.Equal(t, strconv.Itoa(len("secret payload")), rec.Result().Header.Get("Content-Length")) +} + +func TestGetObjectSetsZeroContentLengthForEmptyObject(t *testing.T) { + keks := newTestKEKs(t, "expected encryption key") + version, curKEK := keks.Current() + client := newEncryptedGetObjectClient(t, curKEK, version, []byte{}) + router := Router{keks: keks, log: testLogger()} + req := httptest.NewRequest(http.MethodGet, "/bucket/key", nil) + rec := httptest.NewRecorder() + + router.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + + require.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, "0", rec.Result().Header.Get("Content-Length")) +} + func TestGetObjectFailsWithWrongRouterKEK(t *testing.T) { storedKeks := newTestKEKs(t, "old encryption key") routerKeks := newTestKEKs(t, "new encryption key") From 959c49433c27cf25f7a17110ea2544c4c22c2865 Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Thu, 25 Jun 2026 14:56:09 +0200 Subject: [PATCH 8/9] fix(router): return plaintext ETag on intercepted GetObject MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Intercepted GETs decrypt the body but returned the upstream ETag, which S3 computes over the ciphertext at rest. Clients/SDKs that validate the body against the ETag, or cache by it, saw a mismatch. Override the response ETag with md5(plaintext) — the ETag S3 would have produced for the unencrypted object — computed on the fly from the buffer already in memory, so no stored metadata or migration is needed. Pass-through objects (no DEK tag) keep the upstream ETag. PUT-response and HEAD ETags still reflect the ciphertext and remain a known gap. --- CHANGELOG.md | 10 ++++++ s3proxy/internal/router/object.go | 9 ++++++ s3proxy/internal/router/router_test.go | 44 ++++++++++++++++++++++++++ 3 files changed, 63 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 53d80de..28299b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,16 @@ Two version streams move independently: Previously the proxy streamed decrypted objects without a length, so Go fell back to chunked transfer encoding and s3cmd 2.4.0 downloaded an empty file. Downloads now complete with the correct size. +- **`GetObject` returns the plaintext ETag.** Intercepted GETs decrypt the + body but previously returned the upstream ETag, which S3 computes over the + ciphertext at rest. Clients (or SDKs) that validate the body against the + ETag, or cache by it, saw a mismatch. The proxy now overrides the response + ETag with `md5(plaintext)` — the ETag S3 would have produced for the + unencrypted object — computed on the fly from the buffer already held in + memory, so no stored metadata or migration is needed. Pass-through objects + (no DEK tag) keep the upstream ETag, which already describes the delivered + bytes. PUT-response and HEAD ETags still reflect the ciphertext and remain + a known consistency gap. ### Chart diff --git a/s3proxy/internal/router/object.go b/s3proxy/internal/router/object.go index f5e4cb0..9bcd946 100644 --- a/s3proxy/internal/router/object.go +++ b/s3proxy/internal/router/object.go @@ -9,6 +9,7 @@ package router import ( "context" + "crypto/md5" "encoding/hex" "errors" "fmt" @@ -150,6 +151,14 @@ func (o object) get(w http.ResponseWriter, r *http.Request) { http.Error(w, "failed to decrypt object", http.StatusInternalServerError) return } + + // The upstream ETag is S3's MD5 of the ciphertext at rest, which does not + // match the decrypted body we deliver. Override it with the plaintext MD5 + // so clients that validate or cache by ETag see a consistent value. Single + // part only (multipart is blocked/forwarded), so ETag == hex(md5(content)). + //nolint:gosec // MD5 is not used as a security primitive here; it is the S3 ETag definition. + sum := md5.Sum(plaintext) + w.Header().Set("ETag", hex.EncodeToString(sum[:])) } select { diff --git a/s3proxy/internal/router/router_test.go b/s3proxy/internal/router/router_test.go index a445ceb..bdc906a 100644 --- a/s3proxy/internal/router/router_test.go +++ b/s3proxy/internal/router/router_test.go @@ -8,6 +8,7 @@ package router import ( "bytes" "context" + "crypto/md5" "encoding/hex" "io" "log/slog" @@ -228,6 +229,49 @@ func TestGetObjectSetsZeroContentLengthForEmptyObject(t *testing.T) { assert.Equal(t, "0", rec.Result().Header.Get("Content-Length")) } +func TestGetObjectReturnsPlaintextETag(t *testing.T) { + keks := newTestKEKs(t, "expected encryption key") + version, curKEK := keks.Current() + plaintext := []byte("secret payload") + client := newEncryptedGetObjectClient(t, curKEK, version, plaintext) + // Upstream ETag describes the ciphertext at rest, not the body we deliver. + upstreamETag := "\"deadbeefdeadbeefdeadbeefdeadbeef\"" + client.getObjectOut.ETag = &upstreamETag + router := Router{keks: keks, log: testLogger()} + req := httptest.NewRequest(http.MethodGet, "/bucket/key", nil) + rec := httptest.NewRecorder() + + router.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + + require.Equal(t, http.StatusOK, rec.Code) + want := md5.Sum(plaintext) + assert.Equal(t, hex.EncodeToString(want[:]), rec.Header().Get("ETag")) + assert.NotEqual(t, strings.Trim(upstreamETag, "\""), rec.Header().Get("ETag")) +} + +func TestGetObjectPassThroughKeepsUpstreamETag(t *testing.T) { + // No DEK metadata tag: the object was not proxy-encrypted, so the upstream + // ETag already describes the bytes we deliver and must be left untouched. + body := []byte("plain passthrough") + upstreamETag := "\"deadbeefdeadbeefdeadbeefdeadbeef\"" + client := &recordingS3Client{ + getObjectOut: &s3.GetObjectOutput{ + Body: io.NopCloser(bytes.NewReader(body)), + ContentLength: awsInt64(int64(len(body))), + ETag: &upstreamETag, + }, + } + router := Router{keks: newTestKEKs(t, "expected encryption key"), log: testLogger()} + req := httptest.NewRequest(http.MethodGet, "/bucket/key", nil) + rec := httptest.NewRecorder() + + router.getHandler(req, client, true, "key", "bucket").ServeHTTP(rec, req) + + require.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, body, rec.Body.Bytes()) + assert.Equal(t, strings.Trim(upstreamETag, "\""), rec.Header().Get("ETag")) +} + func TestGetObjectFailsWithWrongRouterKEK(t *testing.T) { storedKeks := newTestKEKs(t, "old encryption key") routerKeks := newTestKEKs(t, "new encryption key") From 05d04245476d4a4d8ee43116284c1c66831f0528 Mon Sep 17 00:00:00 2001 From: Piotr Szafarczyk Date: Thu, 25 Jun 2026 16:02:29 +0200 Subject: [PATCH 9/9] feat(router): report decrypted plaintext size in ListObjects responses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ListObjects/ListObjectsV2 previously reported the at-rest ciphertext size (plaintext + 28-byte AES-GCM-SIV envelope). Intercept bucket-level list requests, subtract the fixed encryption overhead from every , and clamp at 0. Bucket sub-resource GETs (acl, versioning, multipart listings, ?versions, …) are still forwarded unchanged. - cryptoutil: export EncryptionOverhead constant + invariant test - router: bucketOnlyPattern + isListObjects, wired into getHandler - handler: extract forwardUpstream; add handleListObjects (buffer, rewrite on 2xx, fix Content-Length) - listsize: byte-preserving regex rewrite + unit tests - e2e: assert v1 and v2 listings report plaintext size Known limitation: objects not written through the proxy (legacy plaintext, server-side copies, multipart) are reported 28 bytes short / 0 after clamp, since list responses carry no per-object encryption metadata. --- CHANGELOG.md | 9 ++ s3proxy/internal/cryptoutil/cryptoutil.go | 5 + .../internal/cryptoutil/cryptoutil_test.go | 22 ++++ s3proxy/internal/e2e/proxy_e2e_test.go | 16 +++ s3proxy/internal/router/handler.go | 101 +++++++++++++----- s3proxy/internal/router/listsize.go | 39 +++++++ s3proxy/internal/router/listsize_test.go | 62 +++++++++++ s3proxy/internal/router/router.go | 39 ++++++- 8 files changed, 268 insertions(+), 25 deletions(-) create mode 100644 s3proxy/internal/router/listsize.go create mode 100644 s3proxy/internal/router/listsize_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 28299b6..bed05d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,15 @@ Two version streams move independently: (buffers are node-local), assembled object must fit in RAM (~2× peak), `ListParts` unsupported. `Complete` acks only after the upstream store succeeds, inheriting the single-shot path's durability ordering. +- `ListObjects`/`ListObjectsV2` responses now report the **decrypted plaintext + size** of each object instead of the larger at-rest ciphertext size. The proxy + intercepts bucket-level list requests, subtracts the fixed 28-byte encryption + overhead (12-byte nonce + 16-byte GCM tag) from every ``, and clamps at 0. + Bucket sub-resource GETs (acl, versioning, multipart listings, `?versions`, …) + are still forwarded unchanged. + - *Known limitation:* objects **not** written through the proxy (legacy + plaintext, server-side copies, multipart) are reported 28 bytes short (or 0 + after clamp), since a list response carries no per-object encryption metadata. ### Fixed - **s3cmd `get` compatibility**: intercepted `GetObject` responses now diff --git a/s3proxy/internal/cryptoutil/cryptoutil.go b/s3proxy/internal/cryptoutil/cryptoutil.go index 1a0be9b..2e5aabe 100644 --- a/s3proxy/internal/cryptoutil/cryptoutil.go +++ b/s3proxy/internal/cryptoutil/cryptoutil.go @@ -18,6 +18,11 @@ import ( "github.com/tink-crypto/tink-go/v2/subtle/random" ) +// EncryptionOverhead is the fixed number of bytes Encrypt adds to the plaintext: +// a 12-byte AES-GCM-SIV nonce prefix plus a 16-byte authentication tag suffix. +// Ciphertext length is always plaintext length + EncryptionOverhead. +const EncryptionOverhead = 28 + // Encrypt generates a random key to encrypt a plaintext using AES-256-GCM. // The generated key is encrypted using the supplied key encryption key (KEK). // The ciphertext and encrypted data encryption key (DEK) are returned. diff --git a/s3proxy/internal/cryptoutil/cryptoutil_test.go b/s3proxy/internal/cryptoutil/cryptoutil_test.go index 6412452..17ec09d 100644 --- a/s3proxy/internal/cryptoutil/cryptoutil_test.go +++ b/s3proxy/internal/cryptoutil/cryptoutil_test.go @@ -46,3 +46,25 @@ func TestEncryptDecrypt(t *testing.T) { }) } } + +// TestEncryptionOverhead asserts that Encrypt adds exactly EncryptionOverhead bytes to the +// plaintext, regardless of plaintext size. This invariant is what lets the router subtract a +// constant to report decrypted sizes in LIST responses; it fails loudly if the envelope changes. +func TestEncryptionOverhead(t *testing.T) { + for _, size := range []int{0, 1, 28, 29, 1024, 1610862} { + t.Run(fmt.Sprintf("size_%d", size), func(t *testing.T) { + kek := [32]byte{} + _, err := rand.Read(kek[:]) + require.NoError(t, err) + + plaintext := make([]byte, size) + _, err = rand.Read(plaintext) + require.NoError(t, err) + + ciphertext, _, err := Encrypt(plaintext, kek) + require.NoError(t, err) + + assert.Equal(t, size+EncryptionOverhead, len(ciphertext)) + }) + } +} diff --git a/s3proxy/internal/e2e/proxy_e2e_test.go b/s3proxy/internal/e2e/proxy_e2e_test.go index 3c78a85..361082d 100644 --- a/s3proxy/internal/e2e/proxy_e2e_test.go +++ b/s3proxy/internal/e2e/proxy_e2e_test.go @@ -168,6 +168,22 @@ func TestProxyRoundtripAgainstMinio(t *testing.T) { require.NoError(t, got.Body.Close()) assert.Equal(t, plaintext, gotBody) + // Listing through the proxy must report the decrypted plaintext size, not the larger + // ciphertext size that actually sits at rest. Cover both ListObjectsV2 and ListObjects v1. + listV2, err := proxyClient.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{ + Bucket: aws.String(bucket), + }) + require.NoError(t, err) + require.Len(t, listV2.Contents, 1) + assert.EqualValues(t, len(plaintext), *listV2.Contents[0].Size, "ListObjectsV2 must report plaintext size") + + listV1, err := proxyClient.ListObjects(ctx, &awss3.ListObjectsInput{ + Bucket: aws.String(bucket), + }) + require.NoError(t, err) + require.Len(t, listV1.Contents, 1) + assert.EqualValues(t, len(plaintext), *listV1.Contents[0].Size, "ListObjects v1 must report plaintext size") + // Housekeeping endpoints stay reachable while traffic is flowing. healthResp, err := http.Get(proxyURL.String() + "/healthz") require.NoError(t, err) diff --git a/s3proxy/internal/router/handler.go b/s3proxy/internal/router/handler.go index 5871879..0f8a4de 100644 --- a/s3proxy/internal/router/handler.go +++ b/s3proxy/internal/router/handler.go @@ -15,6 +15,7 @@ import ( "io" "log/slog" "net/http" + "strconv" "time" v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4" @@ -138,59 +139,111 @@ func handlePutObject(client s3Client, key string, bucket string, keks cryptoutil } } +// forwardUpstream repackages, signs and sends req to the upstream S3 API, returning the +// upstream response. The caller owns resp.Body and must close it. metrics may be nil. +func forwardUpstream(client *s3.Client, req *http.Request, metrics *monitoring.Metrics) (*http.Response, error) { + newReq, err := repackage(req) + if err != nil { + return nil, fmt.Errorf("repackaging request: %w", err) + } + + cfg := client.GetConfig() + + creds, err := cfg.Credentials.Retrieve(context.TODO()) + if err != nil { + return nil, fmt.Errorf("retrieving aws creds: %w", err) + } + + signer := v4.NewSigner() + + err = signer.SignHTTP(context.TODO(), creds, newReq, newReq.Header.Get("X-Amz-Content-Sha256"), "s3", cfg.Region, time.Now()) + if err != nil { + return nil, fmt.Errorf("signing request: %w", err) + } + + resp, err := forwardHTTPClient.Do(newReq) + if err != nil { + if metrics != nil { + metrics.UpstreamErrors.Inc() + } + return nil, fmt.Errorf("do request: %w", err) + } + + return resp, nil +} + func handleForwards(client *s3.Client, log *slog.Logger, metrics *monitoring.Metrics) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { log.Debug("forwarding", "path", req.URL.Path, "method", req.Method, "host", req.Host) - newReq, err := repackage(req) + resp, err := forwardUpstream(client, req, metrics) if err != nil { - log.Error("failed to repackage request", "error", err) + log.Error("forwarding request", "error", err) http.Error(w, "internal server error", http.StatusInternalServerError) return } + defer func() { + if cerr := resp.Body.Close(); cerr != nil { + log.Error("failed to close upstream response body", "error", cerr) + } + }() - cfg := client.GetConfig() + // Preserve multi-value headers (Set-Cookie, Vary, etc.). + for key, values := range resp.Header { + w.Header()[key] = values + } + w.WriteHeader(resp.StatusCode) - creds, err := cfg.Credentials.Retrieve(context.TODO()) - if err != nil { - log.Error("unable to retrieve aws creds", "error", err) - http.Error(w, "internal server error", http.StatusInternalServerError) - return + if _, err := io.Copy(w, resp.Body); err != nil { + log.Error("failed to stream response", "error", err) + // Headers already written; cannot send error response. } + } +} - signer := v4.NewSigner() +// handleListObjects forwards a ListObjects/ListObjectsV2 request upstream, then rewrites the +// reported object sizes in the XML response to the decrypted plaintext size (ciphertext size +// minus the fixed encryption overhead). List pages are bounded (≤1000 keys), so buffering the +// body is safe. Only successful (2xx) responses are rewritten; everything else passes through. +func handleListObjects(client *s3.Client, log *slog.Logger, metrics *monitoring.Metrics) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + log.Debug("intercepting ListObjects", "path", req.URL.Path, "method", req.Method, "host", req.Host) - err = signer.SignHTTP(context.TODO(), creds, newReq, newReq.Header.Get("X-Amz-Content-Sha256"), "s3", cfg.Region, time.Now()) + resp, err := forwardUpstream(client, req, metrics) if err != nil { - log.Error("failed to sign request", "error", err) + log.Error("forwarding ListObjects request", "error", err) http.Error(w, "internal server error", http.StatusInternalServerError) return } - - resp, err := forwardHTTPClient.Do(newReq) - if err != nil { - log.Error("do request", "error", err) - if metrics != nil { - metrics.UpstreamErrors.Inc() - } - http.Error(w, "do request: "+err.Error(), http.StatusInternalServerError) - return - } defer func() { if cerr := resp.Body.Close(); cerr != nil { log.Error("failed to close upstream response body", "error", cerr) } }() + body, err := io.ReadAll(resp.Body) + if err != nil { + log.Error("reading ListObjects response body", "error", err) + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + + out := body + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + out = rewriteListSizes(body) + } + // Preserve multi-value headers (Set-Cookie, Vary, etc.). for key, values := range resp.Header { w.Header()[key] = values } + // The rewritten body is shorter than the upstream ciphertext sizes, so the upstream + // Content-Length no longer applies — set it to the actual body length. + w.Header().Set("Content-Length", strconv.Itoa(len(out))) w.WriteHeader(resp.StatusCode) - if _, err := io.Copy(w, resp.Body); err != nil { - log.Error("failed to stream response", "error", err) - // Headers already written; cannot send error response. + if _, err := w.Write(out); err != nil { + log.Error("failed to write ListObjects response", "error", err) } } } diff --git a/s3proxy/internal/router/listsize.go b/s3proxy/internal/router/listsize.go new file mode 100644 index 0000000..91044094 --- /dev/null +++ b/s3proxy/internal/router/listsize.go @@ -0,0 +1,39 @@ +/* +Copyright (c) Intrinsec 2024 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package router + +import ( + "fmt" + "regexp" + "strconv" + + "github.com/intrinsec/s3proxy/internal/cryptoutil" +) + +// sizeElementPattern matches a N element. In ListObjects v1/v2 responses this +// element only appears inside , so a targeted byte-preserving substitution avoids +// the namespace/formatting loss of an encoding/xml round-trip that could break S3 clients. +var sizeElementPattern = regexp.MustCompile(`(\d+)`) + +// rewriteListSizes replaces each N in a ListObjects response with +// max(0, N-EncryptionOverhead), reporting the decrypted plaintext size. The clamp at 0 covers +// 0-byte folder markers and objects not written through the proxy (smaller than the overhead). +// Everything outside the matched elements is preserved byte-for-byte. +func rewriteListSizes(body []byte) []byte { + return sizeElementPattern.ReplaceAllFunc(body, func(m []byte) []byte { + n, err := strconv.Atoi(string(sizeElementPattern.FindSubmatch(m)[1])) + if err != nil { + // Unparseable (e.g. overflows int) — leave the element untouched. + return m + } + dec := n - cryptoutil.EncryptionOverhead + if dec < 0 { + dec = 0 + } + return []byte(fmt.Sprintf("%d", dec)) + }) +} diff --git a/s3proxy/internal/router/listsize_test.go b/s3proxy/internal/router/listsize_test.go new file mode 100644 index 0000000..c5922de --- /dev/null +++ b/s3proxy/internal/router/listsize_test.go @@ -0,0 +1,62 @@ +/* +Copyright (c) Intrinsec 2024 + +SPDX-License-Identifier: AGPL-3.0-only +*/ + +package router + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestRewriteListSizes(t *testing.T) { + tests := map[string]struct { + in int + want int + }{ + "zero clamps to zero": {in: 0, want: 0}, + "below overhead clamps to zero": {in: 5, want: 0}, + "exactly overhead clamps to zero": {in: 28, want: 0}, + "one byte plaintext": {in: 29, want: 1}, + "large object": {in: 1610890, want: 1610862}, + } + + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + body := []byte(fmt.Sprintf("%d", tt.in)) + got := rewriteListSizes(body) + assert.Equal(t, fmt.Sprintf("%d", tt.want), string(got)) + }) + } +} + +// TestRewriteListSizesPreservesStructure asserts only values change; surrounding XML — +// including CommonPrefixes (directories, which carry no ) — is byte-for-byte preserved. +func TestRewriteListSizesPreservesStructure(t *testing.T) { + body := []byte(`` + + `` + + `bucket` + + `a.txt1610890"abc"` + + `b.txt28` + + `dir/` + + ``) + + want := []byte(`` + + `` + + `bucket` + + `a.txt1610862"abc"` + + `b.txt0` + + `dir/` + + ``) + + assert.Equal(t, string(want), string(rewriteListSizes(body))) +} + +func TestRewriteListSizesNoSizeElement(t *testing.T) { + body := []byte(`dir/`) + assert.Equal(t, string(body), string(rewriteListSizes(body))) +} diff --git a/s3proxy/internal/router/router.go b/s3proxy/internal/router/router.go index 51733b7..3c0f3a9 100644 --- a/s3proxy/internal/router/router.go +++ b/s3proxy/internal/router/router.go @@ -44,8 +44,41 @@ import ( var ( bucketAndKeyPattern = regexp.MustCompile("/([^/?]+)/(.+)") + // bucketOnlyPattern matches a bucket-level path (no object key), with an optional + // trailing slash. Used to detect ListObjects/ListObjectsV2 requests. + bucketOnlyPattern = regexp.MustCompile("^/([^/?]+)/?$") ) +// listSubResourceMarkers are query parameters that turn a bucket-level GET into a +// sub-resource request (bucket ACL, versioning, multipart listing, …). Those return +// XML that has no , or is out of scope (versions), so they must not be +// rewritten — only a "clean" ListObjects v1/v2 carries object sizes. +var listSubResourceMarkers = []string{ + "acl", "policy", "versioning", "location", "tagging", "lifecycle", "cors", + "website", "replication", "encryption", "object-lock", "accelerate", "analytics", + "intelligent-tiering", "inventory", "metrics", "logging", "notification", + "ownershipControls", "publicAccessBlock", "requestPayment", "uploads", "versions", +} + +// isListObjects reports whether req is a clean ListObjects (v1) or ListObjectsV2 request, +// i.e. a bucket-level GET with no sub-resource marker query parameters. Both v1 (no +// list-type) and v2 (list-type=2) match. +func isListObjects(req *http.Request) bool { + if req.Method != http.MethodGet { + return false + } + if !bucketOnlyPattern.MatchString(req.URL.Path) { + return false + } + query := req.URL.Query() + for _, marker := range listSubResourceMarkers { + if _, ok := query[marker]; ok { + return false + } + } + return true +} + // Router implements the interception logic for the s3proxy. type Router struct { region string @@ -384,8 +417,12 @@ func (r Router) getHandler(req *http.Request, client s3Client, matchingPath bool }) } - // Forward if path doesn't match + // Forward if path doesn't match. Bucket-level ListObjects requests land here (they have + // no object key); intercept them to report decrypted plaintext sizes. if !matchingPath { + if s3Client, ok := client.(*s3.Client); ok && isListObjects(req) { + return handleListObjects(s3Client, r.log, r.metrics) + } return forwardHandler() }