test(remote_storage/s3): assert WriteFile success in mime tests

filer.remote.sync was uploading every object without a Content-Type, so
S3-compatible backends (e.g. Backblaze B2) stored binary/octet-stream
and browsers refused to render HTML, CSS, etc.

Pass entry.Attributes.Mime through to UploadInput.ContentType, leaving
the header unset when no Mime is recorded so the remote keeps its own
default behavior.

seaweed-volume/src/server/handlers.rs

@@ -838,8 +838,6 @@ pub struct ReadQueryParams {
     pub response_content_disposition: Option<String>,
     /// Pretty print JSON response
     pub pretty: Option<String>,
-    /// JSONP callback function name
-    pub callback: Option<String>,
 }
 
 // ============================================================================
@@ -3402,10 +3400,6 @@ fn json_response_with_params<T: Serialize>(
     let is_pretty = params
         .and_then(|params| params.pretty.as_ref())
         .is_some_and(|value| !value.is_empty());
-    let callback = params
-        .and_then(|params| params.callback.as_ref())
-        .filter(|value| !value.is_empty())
-        .cloned();
 
     let json_body = if is_pretty {
         to_pretty_json(body)
@@ -3413,24 +3407,15 @@ fn json_response_with_params<T: Serialize>(
         serde_json::to_string(body).unwrap()
     };
 
-    if let Some(callback) = callback {
-        Response::builder()
-            .status(status)
-            .header(header::CONTENT_TYPE, "application/javascript")
-            .body(Body::from(format!("{}({})", callback, json_body)))
-            .unwrap()
-    } else {
-        Response::builder()
-            .status(status)
-            .header(header::CONTENT_TYPE, "application/json")
-            .body(Body::from(json_body))
-            .unwrap()
-    }
+    Response::builder()
+        .status(status)
+        .header(header::CONTENT_TYPE, "application/json")
+        .header("X-Content-Type-Options", "nosniff")
+        .body(Body::from(json_body))
+        .unwrap()
 }
 
-/// Return a JSON error response with optional query string for pretty/JSONP support.
-/// Supports `?pretty=<any non-empty value>` for pretty-printed JSON and `?callback=fn` for JSONP,
-/// matching Go's writeJsonError behavior.
+/// Return a JSON error response, honoring `?pretty=<any non-empty value>` for pretty-printed JSON.
 pub(super) fn json_error_with_query(
     status: StatusCode,
     msg: impl Into<String>,
@@ -3438,18 +3423,10 @@ pub(super) fn json_error_with_query(
 ) -> Response {
     let body = serde_json::json!({"error": msg.into()});
 
-    let (is_pretty, callback) = if let Some(q) = query {
-        let pretty = q
-            .split('&')
-            .any(|p| p.starts_with("pretty=") && p.len() > "pretty=".len());
-        let cb = q
-            .split('&')
-            .find_map(|p| p.strip_prefix("callback="))
-            .map(|s| s.to_string());
-        (pretty, cb)
-    } else {
-        (false, None)
-    };
+    let is_pretty = query.is_some_and(|q| {
+        q.split('&')
+            .any(|p| p.starts_with("pretty=") && p.len() > "pretty=".len())
+    });
 
     let json_body = if is_pretty {
         to_pretty_json(&body)
@@ -3457,35 +3434,19 @@ pub(super) fn json_error_with_query(
         serde_json::to_string(&body).unwrap()
     };
 
-    if let Some(cb) = callback {
-        let jsonp = format!("{}({})", cb, json_body);
-        Response::builder()
-            .status(status)
-            .header(header::CONTENT_TYPE, "application/javascript")
-            .body(Body::from(jsonp))
-            .unwrap()
-    } else {
-        Response::builder()
-            .status(status)
-            .header(header::CONTENT_TYPE, "application/json")
-            .body(Body::from(json_body))
-            .unwrap()
-    }
+    Response::builder()
+        .status(status)
+        .header(header::CONTENT_TYPE, "application/json")
+        .header("X-Content-Type-Options", "nosniff")
+        .body(Body::from(json_body))
+        .unwrap()
 }
 
-/// Return a JSON response with optional pretty/JSONP support from raw query string.
-/// Matches Go's writeJsonQuiet behavior for write success responses.
+/// Return a JSON response honoring `?pretty=<any non-empty value>` from a raw query string.
 fn json_result_with_query<T: Serialize>(status: StatusCode, body: &T, query: &str) -> Response {
-    let (is_pretty, callback) = {
-        let pretty = query
-            .split('&')
-            .any(|p| p.starts_with("pretty=") && p.len() > "pretty=".len());
-        let cb = query
-            .split('&')
-            .find_map(|p| p.strip_prefix("callback="))
-            .map(|s| s.to_string());
-        (pretty, cb)
-    };
+    let is_pretty = query
+        .split('&')
+        .any(|p| p.starts_with("pretty=") && p.len() > "pretty=".len());
 
     let json_body = if is_pretty {
         to_pretty_json(body)
@@ -3493,20 +3454,12 @@ fn json_result_with_query<T: Serialize>(status: StatusCode, body: &T, query: &st
         serde_json::to_string(body).unwrap()
     };
 
-    if let Some(cb) = callback {
-        let jsonp = format!("{}({})", cb, json_body);
-        Response::builder()
-            .status(status)
-            .header(header::CONTENT_TYPE, "application/javascript")
-            .body(Body::from(jsonp))
-            .unwrap()
-    } else {
-        Response::builder()
-            .status(status)
-            .header(header::CONTENT_TYPE, "application/json")
-            .body(Body::from(json_body))
-            .unwrap()
-    }
+    Response::builder()
+        .status(status)
+        .header(header::CONTENT_TYPE, "application/json")
+        .header("X-Content-Type-Options", "nosniff")
+        .body(Body::from(json_body))
+        .unwrap()
 }
 
 /// Extract JWT token from query param, Authorization header, or Cookie.

test/testutil/ports.go

@@ -11,6 +11,29 @@ import (
 // GrpcPortOffset is the offset weed mini uses to derive gRPC ports from HTTP ports.
 const GrpcPortOffset = 10000
 
+// miniDefaultPorts are the weed mini flag defaults (see weed/command/mini.go).
+// A test only overrides services it uses; unspecified services still bind
+// these defaults, so allocation must avoid handing them out (or any value
+// whose gRPC offset would collide with them).
+var miniDefaultPorts = []int{
+	9333,  // master.port
+	8888,  // filer.port
+	9340,  // volume.port
+	8333,  // s3.port
+	8181,  // s3.port.iceberg
+	7333,  // webdav.port
+	23646, // admin.port
+}
+
+func reservedMiniPorts() map[int]bool {
+	r := make(map[int]bool, len(miniDefaultPorts)*2)
+	for _, p := range miniDefaultPorts {
+		r[p] = true
+		r[p+GrpcPortOffset] = true
+	}
+	return r
+}
+
 // AllocatePorts allocates count unique free ports atomically.
 // All listeners are held open until every port is obtained, preventing
 // the OS from recycling a port between successive allocations.
@@ -63,7 +86,7 @@ func AllocateMiniPorts(count int) ([]int, error) {
 		minPort = 10000
 		maxPort = 55000
 	)
-	reserved := make(map[int]bool)
+	reserved := reservedMiniPorts()
 	ports := make([]int, 0, count)
 	var listeners []net.Listener
 	defer func() {
@@ -135,7 +158,7 @@ func AllocatePortSet(miniCount, regularCount int) (mini []int, regular []int, er
 		minPort = 10000
 		maxPort = 55000
 	)
-	reserved := make(map[int]bool)
+	reserved := reservedMiniPorts()
 	mini = make([]int, 0, miniCount)
 	var listeners []net.Listener
 	defer func() {

test/testutil/ports_test.go

@@ -2,6 +2,29 @@ package testutil
 
 import "testing"
 
+// AllocateMiniPorts must never hand out a port that weed mini will reserve
+// for one of its default services (or that default's gRPC offset). A real
+// failure: Filer was given 33646 (Admin default 23646 + GrpcPortOffset),
+// which mini then refused as "reserved for gRPC calculation".
+func TestAllocateMiniPortsAvoidsMiniDefaults(t *testing.T) {
+	reserved := reservedMiniPorts()
+	for iter := 0; iter < 200; iter++ {
+		ports, err := AllocateMiniPorts(4)
+		if err != nil {
+			t.Fatalf("iter %d: AllocateMiniPorts: %v", iter, err)
+		}
+		for _, p := range ports {
+			if reserved[p] {
+				t.Fatalf("iter %d: allocated port %d is a mini default (or gRPC offset)", iter, p)
+			}
+			if reserved[p+GrpcPortOffset] {
+				t.Fatalf("iter %d: allocated port %d has gRPC offset %d colliding with a mini default",
+					iter, p, p+GrpcPortOffset)
+			}
+		}
+	}
+}
+
 func TestAllocatePortSetNoGrpcCollision(t *testing.T) {
 	// Run a few iterations to catch the OS-recycles-just-closed-port race
 	// that previously hit regular ports when the mini gRPC offset was freed

test/volume_server/http/production_features_test.go

@@ -75,7 +75,7 @@ func TestStatsEndpoints(t *testing.T) {
 	}
 }
 
-func TestStatusPrettyJsonAndJsonp(t *testing.T) {
+func TestStatusPrettyJsonAndCallbackIgnored(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test in short mode")
 	}
@@ -93,29 +93,29 @@ func TestStatusPrettyJsonAndJsonp(t *testing.T) {
 	if len(lines) < 3 {
 		t.Fatalf("/status?pretty=y expected multi-line indented JSON, got %d lines: %s", len(lines), string(prettyBody))
 	}
-	// Verify the body is valid JSON
 	var prettyPayload map[string]interface{}
 	if err := json.Unmarshal(prettyBody, &prettyPayload); err != nil {
 		t.Fatalf("/status?pretty=y is not valid JSON: %v", err)
 	}
 
-	// ?callback=myFunc — expect JSONP wrapping
-	jsonpResp := framework.DoRequest(t, client, mustNewRequest(t, http.MethodGet, cluster.VolumeAdminURL()+"/status?callback=myFunc"))
-	jsonpBody := framework.ReadAllAndClose(t, jsonpResp)
-	if jsonpResp.StatusCode != http.StatusOK {
-		t.Fatalf("/status?callback=myFunc expected 200, got %d", jsonpResp.StatusCode)
+	// ?callback=myFunc — must be ignored; response is plain JSON with nosniff.
+	cbResp := framework.DoRequest(t, client, mustNewRequest(t, http.MethodGet, cluster.VolumeAdminURL()+"/status?callback=myFunc"))
+	cbBody := framework.ReadAllAndClose(t, cbResp)
+	if cbResp.StatusCode != http.StatusOK {
+		t.Fatalf("/status?callback=myFunc expected 200, got %d", cbResp.StatusCode)
 	}
-	bodyStr := string(jsonpBody)
-	if !strings.HasPrefix(bodyStr, "myFunc(") {
-		t.Fatalf("/status?callback=myFunc expected body to start with 'myFunc(', got prefix: %q", bodyStr[:min(len(bodyStr), 30)])
+	if ct := cbResp.Header.Get("Content-Type"); !strings.Contains(ct, "application/json") {
+		t.Fatalf("/status?callback=myFunc expected Content-Type application/json, got %q", ct)
 	}
-	trimmed := strings.TrimRight(bodyStr, "\n; ")
-	if !strings.HasSuffix(trimmed, ")") {
-		t.Fatalf("/status?callback=myFunc expected body to end with ')', got suffix: %q", trimmed[max(0, len(trimmed)-10):])
+	if nosniff := cbResp.Header.Get("X-Content-Type-Options"); nosniff != "nosniff" {
+		t.Fatalf("/status?callback=myFunc expected X-Content-Type-Options nosniff, got %q", nosniff)
 	}
-	// Content-Type should be application/javascript for JSONP
-	if ct := jsonpResp.Header.Get("Content-Type"); !strings.Contains(ct, "javascript") {
-		t.Fatalf("/status?callback=myFunc expected Content-Type containing 'javascript', got %q", ct)
+	if strings.Contains(string(cbBody), "myFunc(") {
+		t.Fatalf("/status?callback=myFunc must not wrap response in callback; body: %q", string(cbBody))
+	}
+	var cbPayload map[string]interface{}
+	if err := json.Unmarshal(cbBody, &cbPayload); err != nil {
+		t.Fatalf("/status?callback=myFunc is not valid JSON: %v", err)
 	}
 }
 

weed/filer/redis2/universal_redis_store_kv.go

@@ -10,7 +10,7 @@ import (
 
 func (store *UniversalRedis2Store) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
 
-	_, err = store.Client.Set(ctx, string(key), value, 0).Result()
+	_, err = store.Client.Set(ctx, store.getKey(string(key)), value, 0).Result()
 
 	if err != nil {
 		return fmt.Errorf("kv put: %w", err)
@@ -21,7 +21,7 @@ func (store *UniversalRedis2Store) KvPut(ctx context.Context, key []byte, value
 
 func (store *UniversalRedis2Store) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
 
-	data, err := store.Client.Get(ctx, string(key)).Result()
+	data, err := store.Client.Get(ctx, store.getKey(string(key))).Result()
 
 	if err == redis.Nil {
 		return nil, filer.ErrKvNotFound
@@ -32,7 +32,7 @@ func (store *UniversalRedis2Store) KvGet(ctx context.Context, key []byte) (value
 
 func (store *UniversalRedis2Store) KvDelete(ctx context.Context, key []byte) (err error) {
 
-	_, err = store.Client.Del(ctx, string(key)).Result()
+	_, err = store.Client.Del(ctx, store.getKey(string(key))).Result()
 
 	if err != nil {
 		return fmt.Errorf("kv delete: %w", err)

weed/remote_storage/s3/s3_storage_client.go

@@ -299,6 +299,9 @@ func (s *s3RemoteStorageClient) WriteFile(loc *remote_pb.RemoteStorageLocation,
 		Body:    reader,
 		Tagging: awsTags,
 	}
+	if entry.Attributes != nil && entry.Attributes.Mime != "" {
+		uploadInput.ContentType = aws.String(entry.Attributes.Mime)
+	}
 	if s.conf.S3StorageClass != "" {
 		uploadInput.StorageClass = aws.String(s.conf.S3StorageClass)
 	}

weed/remote_storage/s3/s3_storage_client_test.go

@@ -1,10 +1,15 @@
 package s3
 
 import (
+	"bytes"
+	"io"
+	"net/http"
+	"strings"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	awss3 "github.com/aws/aws-sdk-go/service/s3"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/remote_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
 	"github.com/stretchr/testify/require"
@@ -65,3 +70,89 @@ func TestS3ErrRemoteObjectNotFoundIsAccessible(t *testing.T) {
 	require.Error(t, remote_storage.ErrRemoteObjectNotFound)
 	require.Equal(t, "remote object not found", remote_storage.ErrRemoteObjectNotFound.Error())
 }
+
+// captureRoundTripper records the PUT request that the s3manager uploader
+// sends, and short-circuits all calls with a 200 so the SDK is satisfied.
+type captureRoundTripper struct {
+	uploadReq *http.Request
+}
+
+func (c *captureRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	if req.Method == http.MethodPut {
+		c.uploadReq = req.Clone(req.Context())
+	}
+	if req.Body != nil {
+		_, _ = io.Copy(io.Discard, req.Body)
+		_ = req.Body.Close()
+	}
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Body:       io.NopCloser(strings.NewReader("")),
+		Header: http.Header{
+			"ETag": []string{"\"etag\""},
+		},
+		Request: req,
+	}, nil
+}
+
+func (c *captureRoundTripper) uploadContentType() string {
+	if c.uploadReq == nil {
+		return ""
+	}
+	return c.uploadReq.Header.Get("Content-Type")
+}
+
+func newCapturingS3Client(t *testing.T) (*s3RemoteStorageClient, *captureRoundTripper) {
+	t.Helper()
+	rt := &captureRoundTripper{}
+	conf := &remote_pb.RemoteConf{
+		Name:             "test",
+		S3Region:         "us-east-1",
+		S3Endpoint:       "https://example.invalid",
+		S3ForcePathStyle: true,
+		S3AccessKey:      "test-key",
+		S3SecretKey:      "test-secret",
+	}
+	httpClient := &http.Client{Transport: rt}
+	rs, err := MakeWithHTTPClient(conf, httpClient)
+	require.NoError(t, err)
+	return rs.(*s3RemoteStorageClient), rt
+}
+
+func TestS3WriteFilePassesMimeAsContentType(t *testing.T) {
+	client, rt := newCapturingS3Client(t)
+	loc := &remote_pb.RemoteStorageLocation{
+		Name:   "test",
+		Bucket: "bucket",
+		Path:   "/dir/test.html",
+	}
+	entry := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Mime: "text/html"},
+	}
+
+	_, err := client.WriteFile(loc, entry, bytes.NewReader([]byte("<html></html>")))
+	require.NoError(t, err)
+
+	require.NotNil(t, rt.uploadReq, "uploader should have issued a PUT")
+	require.Equal(t, "text/html", rt.uploadContentType(), "Content-Type should match entry.Attributes.Mime")
+}
+
+func TestS3WriteFileOmitsContentTypeWhenMimeMissing(t *testing.T) {
+	client, rt := newCapturingS3Client(t)
+	loc := &remote_pb.RemoteStorageLocation{
+		Name:   "test",
+		Bucket: "bucket",
+		Path:   "/dir/test.bin",
+	}
+	entry := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{},
+	}
+
+	_, err := client.WriteFile(loc, entry, bytes.NewReader([]byte("data")))
+	require.NoError(t, err)
+
+	require.NotNil(t, rt.uploadReq, "uploader should have issued a PUT")
+	// When entry.Attributes.Mime is empty we don't force a Content-Type so the
+	// remote can apply its own default rather than getting a misleading one.
+	require.Equal(t, "", rt.uploadContentType())
+}

weed/replication/sink/filersink/fetch_write.go

@@ -120,7 +120,7 @@ func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path str
 
 	fileId, err := fs.fetchAndWrite(sourceChunk, path, sourceMtime)
 	if err != nil {
-		return nil, fmt.Errorf("copy %s: %v", sourceChunk.GetFileIdString(), err)
+		return nil, fmt.Errorf("copy %s: %w", sourceChunk.GetFileIdString(), err)
 	}
 
 	return &filer_pb.FileChunk{
@@ -292,6 +292,10 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string,
 			fullData = data
 		}
 
+		if err := validateReplicatedReadSize(sourceChunk, len(fullData)); err != nil {
+			return err
+		}
+
 		transferStatus.mu.Lock()
 		transferStatus.BytesReceived = int64(len(fullData))
 		transferStatus.Status = "uploading"
@@ -335,6 +339,14 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string,
 		fileId = currentFileId
 		return nil
 	}, func(retryErr error) (shouldContinue bool) {
+		if errors.Is(retryErr, errChunkSizeMismatch) {
+			glog.V(0).Infof("permanent size mismatch replicating %s for %s: %v",
+				sourceChunk.GetFileIdString(), path, retryErr)
+			transferStatus.mu.Lock()
+			transferStatus.LastErr = retryErr.Error()
+			transferStatus.mu.Unlock()
+			return false
+		}
 		if fs.hasSourceNewerVersion(path, sourceMtime) {
 			glog.V(1).Infof("skip retrying stale source %s for %s: %v", sourceChunk.GetFileIdString(), path, retryErr)
 			return false
@@ -388,6 +400,18 @@ func isEofError(err error) bool {
 	return errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF)
 }
 
+// errChunkSizeMismatch is a permanent (non-retriable) replication failure.
+var errChunkSizeMismatch = errors.New("chunk size mismatch")
+
+func validateReplicatedReadSize(sourceChunk *filer_pb.FileChunk, readSize int) error {
+	if uint64(readSize) != sourceChunk.Size {
+		return fmt.Errorf("%w: read %s got %d bytes, source metadata says %d",
+			errChunkSizeMismatch, sourceChunk.GetFileIdString(),
+			readSize, sourceChunk.Size)
+	}
+	return nil
+}
+
 func (fs *FilerSink) buildUploadUrl(host, fileId string) string {
 	if fs.writeChunkByFiler {
 		return fmt.Sprintf("http://%s/?proxyChunkId=%s", fs.address, fileId)

weed/replication/sink/filersink/fetch_write_test.go

@@ -1,12 +1,27 @@
 package filersink
 
 import (
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/replication/source"
 	"github.com/seaweedfs/seaweedfs/weed/util"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
+func TestMain(m *testing.M) {
+	util_http.InitGlobalHttpClient()
+	os.Exit(m.Run())
+}
+
 func TestTargetPathToSourcePath(t *testing.T) {
 	tests := []struct {
 		name       string
@@ -77,3 +92,191 @@ func TestTargetPathToSourcePath(t *testing.T) {
 		})
 	}
 }
+
+// FilerSink must reject chunks whose received byte count disagrees with the
+// source filer metadata, instead of silently writing 0-byte needles with the
+// source size in the destination metadata.
+func TestValidateReplicatedChunkSize(t *testing.T) {
+	const fid = "74,047d16a94aa581"
+
+	tests := []struct {
+		name         string
+		expectedSize uint64
+		readSize     int
+		wantErr      bool
+	}{
+		{
+			name:         "healthy",
+			expectedSize: 5171,
+			readSize:     5171,
+			wantErr:      false,
+		},
+		{
+			name:         "legitimately empty file",
+			expectedSize: 0,
+			readSize:     0,
+			wantErr:      false,
+		},
+		{
+			name:         "zero-byte read for non-empty source",
+			expectedSize: 5171,
+			readSize:     0,
+			wantErr:      true,
+		},
+		{
+			name:         "short read",
+			expectedSize: 5171,
+			readSize:     100,
+			wantErr:      true,
+		},
+		{
+			name:         "over-read (server returned more than metadata)",
+			expectedSize: 5171,
+			readSize:     8192,
+			wantErr:      true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			chunk := &filer_pb.FileChunk{FileId: fid, Size: tc.expectedSize}
+
+			gotErr := validateReplicatedReadSize(chunk, tc.readSize)
+
+			if tc.wantErr {
+				if gotErr == nil {
+					t.Fatalf("expected error, got nil (read=%d expected=%d)",
+						tc.readSize, tc.expectedSize)
+				}
+				if !errors.Is(gotErr, errChunkSizeMismatch) {
+					t.Fatalf("expected errChunkSizeMismatch, got %v", gotErr)
+				}
+				if !strings.Contains(gotErr.Error(), fid) {
+					t.Fatalf("error %q does not mention chunk id %q", gotErr, fid)
+				}
+				return
+			}
+			if gotErr != nil {
+				t.Fatalf("unexpected read-size error: %v", gotErr)
+			}
+		})
+	}
+}
+
+// End-to-end regression :
+// a source volume that responds 200 OK with Content-Length: 0 
+// for a chunk that filer metadata claims is 5171 bytes must be rejected 
+// by fetchAndWrite with a (non-retriable) size mismatch error, 
+// instead of being silently propagated to the destination as a 0-byte needle.
+func TestFetchAndWriteRejectsZeroByteSource(t *testing.T) {
+	const fid = "74,047d16a94aa581"
+	const expectedSize uint64 = 5171
+
+	// Shorten retry backoff so a fail-fast test that briefly enters the retry
+	// loop doesn't pay the production 1s+ wait. Scoped to this test so any
+	// future test in the package keeps the production constant.
+	prevRetryWaitTime := util.RetryWaitTime
+	util.RetryWaitTime = 100 * time.Millisecond
+	t.Cleanup(func() { util.RetryWaitTime = prevRetryWaitTime })
+
+	var hits atomic.Int32
+	sourceServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits.Add(1)
+		w.Header().Set("Content-Type", "application/octet-stream")
+		w.WriteHeader(http.StatusOK)
+		// Intentionally write no body — mimic the buggy volume response.
+	}))
+	defer sourceServer.Close()
+
+	serverAddr := strings.TrimPrefix(sourceServer.URL, "http://")
+
+	filerSrc := &source.FilerSource{}
+	if err := filerSrc.DoInitialize(serverAddr, serverAddr, "/", true); err != nil {
+		t.Fatalf("filerSource.DoInitialize: %v", err)
+	}
+
+	fs := &FilerSink{
+		filerSource: filerSrc,
+		address:     serverAddr,
+		dir:         "/dst",
+		executor:    util.NewLimitedConcurrentExecutor(1),
+	}
+	fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient))
+
+	sourceChunk := &filer_pb.FileChunk{
+		FileId: fid,
+		Size:   expectedSize,
+	}
+
+	done := make(chan struct {
+		fileId string
+		err    error
+	}, 1)
+	go func() {
+		gotFileId, gotErr := fs.fetchAndWrite(sourceChunk, "/dst/index.bin", 0)
+		done <- struct {
+			fileId string
+			err    error
+		}{gotFileId, gotErr}
+	}()
+
+	select {
+	case result := <-done:
+		if result.err == nil {
+			t.Fatalf("expected size mismatch error, got nil (fileId=%q)", result.fileId)
+		}
+		if !errors.Is(result.err, errChunkSizeMismatch) {
+			t.Fatalf("expected errChunkSizeMismatch, got %v", result.err)
+		}
+		if !strings.Contains(result.err.Error(), "5171") {
+			t.Fatalf("error %q does not mention expected size 5171", result.err)
+		}
+		if !strings.Contains(result.err.Error(), fid) {
+			t.Fatalf("error %q does not mention chunk id %q", result.err, fid)
+		}
+		if h := hits.Load(); h != 1 {
+			t.Fatalf("expected exactly 1 source hit (fail-fast), got %d", h)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatalf("fetchAndWrite did not return within 5s (retry loop not aborted on size mismatch); hits=%d", hits.Load())
+	}
+}
+
+// Lock in that the errChunkSizeMismatch sentinel survives the wrap in
+// replicateOneChunk + pass-through in util.Retry, so filer_sink.go's
+// errors.Is check actually fires.
+func TestReplicateChunksPreservesSizeMismatchSentinel(t *testing.T) {
+	const fid = "74,047d16a94aa581"
+	const expectedSize uint64 = 5171
+
+	sourceServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/octet-stream")
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer sourceServer.Close()
+
+	serverAddr := strings.TrimPrefix(sourceServer.URL, "http://")
+
+	filerSrc := &source.FilerSource{}
+	if err := filerSrc.DoInitialize(serverAddr, serverAddr, "/", true); err != nil {
+		t.Fatalf("filerSource.DoInitialize: %v", err)
+	}
+
+	fs := &FilerSink{
+		filerSource: filerSrc,
+		address:     serverAddr,
+		dir:         "/dst",
+		executor:    util.NewLimitedConcurrentExecutor(1),
+	}
+	fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient))
+
+	sourceChunks := []*filer_pb.FileChunk{{FileId: fid, Size: expectedSize}}
+
+	_, err := fs.replicateChunks(nil, sourceChunks, "/dst/index.bin", 0)
+	if err == nil {
+		t.Fatal("expected error from replicateChunks, got nil")
+	}
+	if !errors.Is(err, errChunkSizeMismatch) {
+		t.Fatalf("error chain broken: errors.Is(err, errChunkSizeMismatch) = false; got %v", err)
+	}
+}

weed/replication/sink/filersink/filer_sink.go

@@ -2,6 +2,7 @@ package filersink
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"sync"
@@ -191,7 +192,12 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [
 		replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtime(entry))
 
 		if err != nil {
-			// only warning here since the source chunk may have been deleted already
+			// Don't swallow size-mismatch: source bytes disagree with source
+			// metadata, so committing would propagate corruption silently.
+			if errors.Is(err, errChunkSizeMismatch) {
+				glog.Errorf("refuse to replicate entry with corrupt chunk %s: %v", key, err)
+				return err
+			}
 			glog.Warningf("replicate entry chunks %s: %v", key, err)
 			return nil
 		}
@@ -259,8 +265,8 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 		// this usually happens when the messages are not ordered
 		glog.V(2).Infof("late updates %s", key)
 	} else {
-		// find out what changed
-		deletedChunks, newChunks, err := compareChunks(context.Background(), filer.LookupFn(fs), oldEntry, newEntry)
+		// source-side chunks resolve via source filer; sink volume IDs may collide.
+		deletedChunks, newChunks, err := compareChunks(context.Background(), filer.LookupFn(fs.filerSource), oldEntry, newEntry)
 		if err != nil {
 			return true, fmt.Errorf("replicate %s compare chunks error: %v", key, err)
 		}
@@ -274,6 +280,10 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 		// replicate the chunks that are new in the source
 		replicatedChunks, err := fs.replicateChunks(context.Background(), newChunks, key, getEntryMtime(newEntry))
 		if err != nil {
+			if errors.Is(err, errChunkSizeMismatch) {
+				glog.Errorf("refuse to replicate entry with corrupt chunk %s: %v", key, err)
+				return true, err
+			}
 			glog.Warningf("replicate entry chunks %s: %v", key, err)
 			return true, nil
 		}

weed/s3api/iceberg/path_validation.go

@@ -0,0 +1,70 @@
+package iceberg
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+)
+
+// validateRequestPath rejects Iceberg REST requests whose captured
+// {prefix}/{namespace}/{table} mux vars would produce a parent-directory
+// traversal when joined into a filer path. The iceberg router runs with
+// SkipClean(true), so `..` survives routing; downstream path.Join calls
+// (stageCreateMarkerDir, location builders, etc.) then collapse it and
+// escape the table-bucket directory.
+//
+// {prefix} maps to a table-bucket name; {table} is a single path segment;
+// {namespace} is unit-separator (0x1F) joined parts that get flattened into
+// a single dotted name for the on-disk layout — each part is validated
+// individually.
+func validateRequestPath(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		// Use the comma-ok form so vars only checked when the matched route
+		// actually captures them; when captured, an empty value is itself a
+		// rejection because downstream path.Join would collapse it.
+		if prefix, ok := vars["prefix"]; ok {
+			if prefix == "" || !s3_constants.IsValidBucketName(prefix) {
+				writeError(w, http.StatusBadRequest, "BadRequest", "invalid prefix")
+				return
+			}
+		}
+		if table, ok := vars["table"]; ok {
+			if table == "" || !isValidNameSegment(table) {
+				writeError(w, http.StatusBadRequest, "BadRequest", "invalid table name")
+				return
+			}
+		}
+		if ns, ok := vars["namespace"]; ok {
+			if ns == "" {
+				writeError(w, http.StatusBadRequest, "BadRequest", "invalid namespace")
+				return
+			}
+			// Reject leading/trailing/consecutive unit separators so distinct
+			// inputs cannot collapse to the same parsed namespace via
+			// parseNamespace's empty-part filter.
+			for _, part := range strings.Split(ns, "\x1F") {
+				if part == "" || !isValidNameSegment(part) {
+					writeError(w, http.StatusBadRequest, "BadRequest", "invalid namespace")
+					return
+				}
+			}
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
+// isValidNameSegment rejects a single path-segment value (bucket prefix slot,
+// table name, or one namespace part) that would be unsafe to embed in a filer
+// path: `.`, `..`, embedded slash/backslash, or NUL.
+func isValidNameSegment(s string) bool {
+	if s == "" {
+		return true
+	}
+	if s == "." || s == ".." {
+		return false
+	}
+	return !strings.ContainsAny(s, "/\\\x00")
+}

weed/s3api/iceberg/path_validation_test.go

@@ -0,0 +1,116 @@
+package iceberg
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gorilla/mux"
+)
+
+func TestValidateRequestPath_RejectsTraversal(t *testing.T) {
+	tests := []struct {
+		name     string
+		rawPath  string
+		wantCode int
+	}{
+		{"clean namespace+table passes", "/v1/namespaces/sales/tables/orders", http.StatusOK},
+		{"clean prefixed passes", "/v1/wh/namespaces/sales/tables/orders", http.StatusOK},
+		{"clean namespace only passes", "/v1/namespaces/sales", http.StatusOK},
+
+		// SkipClean(true) means raw `..` survives routing — these are the
+		// realistic traversal shapes the middleware must catch.
+		{"dotdot as prefix var rejected", "/v1/../namespaces/sales", http.StatusBadRequest},
+		{"dotdot as namespace var rejected", "/v1/namespaces/..", http.StatusBadRequest},
+		{"dotdot as namespace var prefixed rejected", "/v1/wh/namespaces/..", http.StatusBadRequest},
+		{"dotdot as table var rejected", "/v1/namespaces/sales/tables/..", http.StatusBadRequest},
+		{"dot as table var rejected", "/v1/namespaces/sales/tables/.", http.StatusBadRequest},
+		// Iceberg clients send the 0x1F unit separator percent-encoded; mux
+		// decodes it before the middleware sees the namespace var.
+		{"unit-sep namespace with dotdot part rejected", "/v1/namespaces/sales%1F..%1Fevil", http.StatusBadRequest},
+		{"leading unit-sep namespace rejected", "/v1/namespaces/%1Fsales", http.StatusBadRequest},
+		{"trailing unit-sep namespace rejected", "/v1/namespaces/sales%1F", http.StatusBadRequest},
+		{"consecutive unit-sep namespace rejected", "/v1/namespaces/sales%1F%1Fevil", http.StatusBadRequest},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			router := mux.NewRouter().SkipClean(true)
+			router.Use(validateRequestPath)
+			handlerCalled := false
+			pass := func(w http.ResponseWriter, r *http.Request) {
+				handlerCalled = true
+				w.WriteHeader(http.StatusOK)
+			}
+			router.HandleFunc("/v1/namespaces/{namespace}", pass)
+			router.HandleFunc("/v1/namespaces/{namespace}/tables/{table}", pass)
+			router.HandleFunc("/v1/{prefix}/namespaces/{namespace}", pass)
+			router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables/{table}", pass)
+
+			req := httptest.NewRequest(http.MethodGet, tt.rawPath, nil)
+			rr := httptest.NewRecorder()
+			router.ServeHTTP(rr, req)
+
+			if rr.Code != tt.wantCode {
+				t.Fatalf("path %q: got status %d, want %d (body=%q)", tt.rawPath, rr.Code, tt.wantCode, rr.Body.String())
+			}
+			if tt.wantCode == http.StatusBadRequest && handlerCalled {
+				t.Fatalf("path %q: inner handler reached despite rejection", tt.rawPath)
+			}
+		})
+	}
+}
+
+// Defense-in-depth: if a future route or middleware ever leaves one of the
+// captured vars empty, the middleware must still reject the request. The
+// default mux regex won't normally allow this.
+func TestValidateRequestPath_RejectsEmptyCapturedVars(t *testing.T) {
+	tests := []struct {
+		name string
+		vars map[string]string
+	}{
+		{"empty prefix", map[string]string{"prefix": "", "namespace": "ns"}},
+		{"empty table", map[string]string{"namespace": "ns", "table": ""}},
+		{"empty namespace", map[string]string{"namespace": ""}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handlerCalled := false
+			h := validateRequestPath(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				handlerCalled = true
+			}))
+			req := mux.SetURLVars(httptest.NewRequest(http.MethodGet, "/", nil), tt.vars)
+			rr := httptest.NewRecorder()
+			h.ServeHTTP(rr, req)
+			if handlerCalled {
+				t.Fatalf("vars %v: inner handler reached despite empty capture", tt.vars)
+			}
+		})
+	}
+}
+
+func TestIsValidNameSegment(t *testing.T) {
+	tests := []struct {
+		name  string
+		input string
+		want  bool
+	}{
+		{"empty ok", "", true},
+		{"plain", "orders", true},
+		{"with dot inside", "my.table", true},
+		{"hidden", ".hidden", true},
+
+		{"bare dot", ".", false},
+		{"bare dotdot", "..", false},
+		{"contains slash", "foo/bar", false},
+		{"contains backslash", "foo\\bar", false},
+		{"contains nul", "foo\x00bar", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isValidNameSegment(tt.input); got != tt.want {
+				t.Errorf("isValidNameSegment(%q) = %v, want %v", tt.input, got, tt.want)
+			}
+		})
+	}
+}

weed/s3api/iceberg/server.go

@@ -71,6 +71,13 @@ func (s *Server) RegisterRoutes(router *mux.Router) {
 	// Add middleware to log all requests/responses
 	router.Use(loggingMiddleware)
 
+	// Reject `..`/`.`/NUL in {prefix}/{namespace}/{table} vars before any
+	// handler runs. The router uses SkipClean(true), so traversal segments
+	// would otherwise reach path.Join in stage-marker / location builders.
+	// Registered after loggingMiddleware so rejected requests still get
+	// audit-logged.
+	router.Use(validateRequestPath)
+
 	// Configuration endpoint - no auth needed for config
 	router.HandleFunc("/v1/config", s.handleConfig).Methods(http.MethodGet)
 

weed/s3api/s3_constants/header.go

@@ -177,6 +177,41 @@ func GetBucketAndObject(r *http.Request) (bucket, object string) {
 	return
 }
 
+// IsValidObjectKey rejects S3 object keys that — after normalization
+// (backslash→slash, slash collapse) — contain a `.` or `..` path segment, or
+// embed a NUL byte. Such keys are collapsed by filepath.Join inside the filer
+// and would escape the bucket directory, so they must not reach the gRPC layer.
+// Gorilla mux URL-decodes captured vars before this runs, so `%2e%2e` is
+// already `..` here.
+func IsValidObjectKey(object string) bool {
+	if object == "" {
+		return true
+	}
+	if strings.ContainsRune(object, '\x00') {
+		return false
+	}
+	object = strings.ReplaceAll(object, "\\", "/")
+	for _, seg := range strings.Split(object, "/") {
+		if seg == "." || seg == ".." {
+			return false
+		}
+	}
+	return true
+}
+
+// IsValidBucketName rejects bucket names captured from the URL path that are
+// unsafe to use in filer path construction (`.`, `..`, contain `/` or `\`, or
+// embed NUL). This is a path-safety check, not a full S3 naming-rule check.
+func IsValidBucketName(bucket string) bool {
+	if bucket == "" {
+		return true
+	}
+	if bucket == "." || bucket == ".." {
+		return false
+	}
+	return !strings.ContainsAny(bucket, "/\\\x00")
+}
+
 // NormalizeObjectKey normalizes object keys by removing duplicate slashes and converting backslashes.
 // This normalizes keys from various sources (URL path, form values, etc.) to a consistent format.
 // It also converts Windows-style backslashes to forward slashes for cross-platform compatibility.

weed/s3api/s3_constants/header_test.go

@@ -89,6 +89,65 @@ func TestNormalizeObjectKey(t *testing.T) {
 	}
 }
 
+func TestIsValidObjectKey(t *testing.T) {
+	tests := []struct {
+		name  string
+		input string
+		want  bool
+	}{
+		{"empty", "", true},
+		{"plain", "folder/file.txt", true},
+		{"leading slash", "/folder/file.txt", true},
+		{"trailing slash", "folder/", true},
+		{"hidden file ok", ".hidden", true},
+		{"dotdot in name ok", "..hidden", true},
+		{"double dots inside name", "foo..bar/baz", true},
+
+		{"bare dotdot", "..", false},
+		{"bare dot", ".", false},
+		{"leading dotdot segment", "../evil-bucket/test.txt", false},
+		{"leading dot-slash", "./evil/test.txt", false},
+		{"nested dotdot segment", "good/../evil/test.txt", false},
+		{"trailing dotdot segment", "good/..", false},
+		{"backslash dotdot", "..\\evil\\test.txt", false},
+		{"mixed-slash dotdot", "good\\..\\evil/test.txt", false},
+		{"dotdot after duplicate slash", "good//../evil", false},
+		{"nul byte", "foo\x00bar", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := IsValidObjectKey(tt.input); got != tt.want {
+				t.Errorf("IsValidObjectKey(%q) = %v, want %v", tt.input, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestIsValidBucketName(t *testing.T) {
+	tests := []struct {
+		name  string
+		input string
+		want  bool
+	}{
+		{"empty ok", "", true},
+		{"plain", "my-bucket", true},
+		{"name containing dots", "my.bucket.name", true},
+
+		{"bare dot", ".", false},
+		{"bare dotdot", "..", false},
+		{"with slash", "evil/bucket", false},
+		{"with backslash", "evil\\bucket", false},
+		{"with nul", "evil\x00bucket", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := IsValidBucketName(tt.input); got != tt.want {
+				t.Errorf("IsValidBucketName(%q) = %v, want %v", tt.input, got, tt.want)
+			}
+		})
+	}
+}
+
 func TestRemoveDuplicateSlashes(t *testing.T) {
 	tests := []struct {
 		name     string

weed/s3api/s3api_path_validation.go

@@ -0,0 +1,38 @@
+package s3api
+
+import (
+	"net/http"
+
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+)
+
+// validateRequestPath rejects requests whose captured {bucket}/{object} mux
+// vars would normalize to a parent-directory traversal once joined into a
+// filer path. The router runs with mux.NewRouter().SkipClean(true), so
+// segments like `..` survive routing; the filer's util.JoinPath later collapses
+// them via filepath.Join. Without this guard, `GET /bucket-A/../evil-bucket/k`
+// matches as bucket=bucket-A, object=../evil-bucket/k, the filer resolves the
+// read against evil-bucket, while IAM authorizes against bucket-A.
+func validateRequestPath(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		vars := mux.Vars(r)
+		// When a var is in the matched route it must be non-empty: an empty
+		// bucket would let downstream path.Join collapse it and let the object
+		// key pick the bucket.
+		if bucket, ok := vars["bucket"]; ok {
+			if bucket == "" || !s3_constants.IsValidBucketName(bucket) {
+				s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
+				return
+			}
+		}
+		if object, ok := vars["object"]; ok {
+			if object == "" || !s3_constants.IsValidObjectKey(object) {
+				s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
+				return
+			}
+		}
+		next.ServeHTTP(w, r)
+	})
+}

weed/s3api/s3api_path_validation_test.go

@@ -0,0 +1,87 @@
+package s3api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/gorilla/mux"
+)
+
+func TestValidateRequestPath_RejectsTraversal(t *testing.T) {
+	tests := []struct {
+		name string
+		// rawPath is sent as the Request-URI; net/http.NewRequest does not
+		// rewrite the path, so `..` segments survive into mux when the router
+		// is built with SkipClean(true) — matching the production setup in
+		// weed/command/s3.go.
+		rawPath  string
+		wantCode int
+	}{
+		{"clean path passes", "/bucket-a/folder/file.txt", http.StatusOK},
+		{"bucket only passes", "/bucket-a", http.StatusOK},
+		{"trailing slash passes", "/bucket-a/folder/", http.StatusOK},
+
+		{"leading dotdot rejected", "/bucket-a/../evil-bucket/test.txt", http.StatusBadRequest},
+		{"nested dotdot rejected", "/bucket-a/good/../evil/test.txt", http.StatusBadRequest},
+		{"backslash dotdot rejected", "/bucket-a/..\\evil\\test.txt", http.StatusBadRequest},
+		{"percent-encoded dotdot rejected", "/bucket-a/%2e%2e/evil/test.txt", http.StatusBadRequest},
+		{"bare dot object rejected", "/bucket-a/./evil/test.txt", http.StatusBadRequest},
+		{"dotdot bucket rejected", "/../buckets/evil", http.StatusBadRequest},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			router := mux.NewRouter().SkipClean(true)
+			sub := router.PathPrefix("/{bucket}").Subrouter()
+			sub.Use(validateRequestPath)
+			handlerCalled := false
+			pass := func(w http.ResponseWriter, r *http.Request) {
+				handlerCalled = true
+				w.WriteHeader(http.StatusOK)
+			}
+			// Mirror the production routes: /{bucket}/{object:(?s).+} for
+			// object-scoped requests, bare /{bucket} for bucket-scoped ones.
+			sub.Path("/{object:(?s).+}").HandlerFunc(pass)
+			sub.Path("").HandlerFunc(pass)
+
+			req := httptest.NewRequest(http.MethodGet, tt.rawPath, nil)
+			rr := httptest.NewRecorder()
+			router.ServeHTTP(rr, req)
+
+			if rr.Code != tt.wantCode {
+				t.Fatalf("path %q: got status %d, want %d (body=%q)", tt.rawPath, rr.Code, tt.wantCode, rr.Body.String())
+			}
+			if tt.wantCode == http.StatusBadRequest && handlerCalled {
+				t.Fatalf("path %q: inner handler reached despite rejection", tt.rawPath)
+			}
+		})
+	}
+}
+
+// Defense-in-depth: a future router or middleware that captures the {bucket}
+// or {object} mux var as an empty string must still be rejected, even though
+// mux's default `[^/]+` regex won't match an empty segment from a real URL.
+func TestValidateRequestPath_RejectsEmptyCapturedVars(t *testing.T) {
+	tests := []struct {
+		name string
+		vars map[string]string
+	}{
+		{"empty bucket", map[string]string{"bucket": "", "object": "key"}},
+		{"empty object", map[string]string{"bucket": "bucket-a", "object": ""}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handlerCalled := false
+			h := validateRequestPath(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				handlerCalled = true
+			}))
+			req := mux.SetURLVars(httptest.NewRequest(http.MethodGet, "/", nil), tt.vars)
+			rr := httptest.NewRecorder()
+			h.ServeHTTP(rr, req)
+			if handlerCalled {
+				t.Fatalf("vars %v: inner handler reached despite empty capture", tt.vars)
+			}
+		})
+	}
+}

weed/s3api/s3api_server.go

@@ -735,6 +735,11 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
 	corsMiddleware := s3a.getCORSMiddleware()
 
 	for _, bucket := range routers {
+		// Reject `..`/`.`/NUL in {bucket} or {object} vars before any handler
+		// runs. SkipClean(true) keeps `..` in the matched path; the filer would
+		// otherwise collapse it via filepath.Join and cross bucket boundaries.
+		bucket.Use(validateRequestPath)
+
 		// Apply CORS middleware to bucket routers for automatic CORS header handling
 		bucket.Use(corsMiddleware.Handler)
 

weed/server/common.go

@@ -110,32 +110,10 @@ func writeJson(w http.ResponseWriter, r *http.Request, httpStatus int, obj inter
 			r.Method, r.URL.String(), httpStatus, string(bytes))
 	}
 
-	callback := r.FormValue("callback")
-	if callback == "" {
-		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(httpStatus)
-		if httpStatus == http.StatusNotModified {
-			return
-		}
-		_, err = w.Write(bytes)
-	} else {
-		w.Header().Set("Content-Type", "application/javascript")
-		w.WriteHeader(httpStatus)
-		if httpStatus == http.StatusNotModified {
-			return
-		}
-		if _, err = w.Write([]uint8(callback)); err != nil {
-			return
-		}
-		if _, err = w.Write([]uint8("(")); err != nil {
-			return
-		}
-		fmt.Fprint(w, string(bytes))
-		if _, err = w.Write([]uint8(")")); err != nil {
-			return
-		}
-	}
-
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("X-Content-Type-Options", "nosniff")
+	w.WriteHeader(httpStatus)
+	_, err = w.Write(bytes)
 	return
 }
 

weed/server/common_test.go

@@ -1,6 +1,8 @@
 package weed_server
 
 import (
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 )
@@ -29,3 +31,33 @@ func TestParseURL(t *testing.T) {
 		}
 	}
 }
+
+func TestWriteJsonNoJSONP(t *testing.T) {
+	// callback= must be ignored; response is always application/json with nosniff.
+	cases := []string{"", "myCb", "<script>alert(1)</script>"}
+	for _, cb := range cases {
+		t.Run("callback="+cb, func(t *testing.T) {
+			url := "/x"
+			if cb != "" {
+				url += "?callback=" + cb
+			}
+			r := httptest.NewRequest(http.MethodGet, url, nil)
+			w := httptest.NewRecorder()
+			if err := writeJson(w, r, http.StatusOK, map[string]string{"k": "v"}); err != nil {
+				t.Fatalf("writeJson: %v", err)
+			}
+			if w.Code != http.StatusOK {
+				t.Errorf("status: got %d want 200", w.Code)
+			}
+			if got := w.Header().Get("Content-Type"); got != "application/json" {
+				t.Errorf("Content-Type: got %q want application/json", got)
+			}
+			if got := w.Header().Get("X-Content-Type-Options"); got != "nosniff" {
+				t.Errorf("X-Content-Type-Options: got %q want nosniff", got)
+			}
+			if got := w.Body.String(); got != `{"k":"v"}` {
+				t.Errorf("body: got %q want %q", got, `{"k":"v"}`)
+			}
+		})
+	}
+}

weed/server/master_grpc_server_volume.go

@@ -152,9 +152,10 @@ func (ms *MasterServer) ProcessGrowRequest() {
 			// we have lock called inside vg
 			glog.V(0).Infof("volume grow %+v", req)
 			go func(req *topology.VolumeGrowRequest, vl *topology.VolumeLayout) {
+				// defer so a panic can't strand growRequest.
+				defer filter.Delete(req)
+				defer vl.DoneGrowRequest()
 				ms.DoAutomaticVolumeGrow(req)
-				vl.DoneGrowRequest()
-				filter.Delete(req)
 			}(req, vl)
 		}
 	}()

weed/storage/volume_loading.go

@@ -279,9 +279,14 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		// (issue #8928). The check piggybacks on MaxNeedleEnd, which the load
 		// walks below populate without a second linear scan.
 
+		// Loaders can return a typed-nil pointer with err set; assigning that
+		// to v.nm yields a non-nil interface over a nil receiver. Clear v.nm
+		// and close indexFile so the defer cleanup keys off v.nm cleanly.
 		if v.noWriteOrDelete || v.noWriteCanDelete {
 			if v.nm, err = NewSortedFileNeedleMap(v.IndexFileName(), indexFile, v.Version()); err != nil {
 				glog.V(0).Infof("loading sorted db %s error: %v", v.FileName(".sdx"), err)
+				v.nm = nil
+				indexFile.Close()
 			}
 		} else {
 			switch needleMapKind {
@@ -293,6 +298,8 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 					glog.V(2).Infoln("loading memory index", v.FileName(".idx"), "to memory")
 					if v.nm, err = LoadCompactNeedleMap(indexFile, v.Version()); err != nil {
 						glog.V(0).Infof("loading index %s to memory error: %v", v.FileName(".idx"), err)
+						v.nm = nil
+						indexFile.Close()
 					}
 				}
 			case NeedleMapLevelDb:
@@ -309,6 +316,8 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 					glog.V(0).Infoln("loading leveldb index", v.FileName(".ldb"))
 					if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
 						glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
+						v.nm = nil
+						indexFile.Close()
 					}
 				}
 			case NeedleMapLevelDbMedium:
@@ -325,6 +334,8 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 					glog.V(0).Infoln("loading leveldb medium index", v.FileName(".ldb"))
 					if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
 						glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
+						v.nm = nil
+						indexFile.Close()
 					}
 				}
 			case NeedleMapLevelDbLarge:
@@ -341,6 +352,8 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 					glog.V(0).Infoln("loading leveldb large index", v.FileName(".ldb"))
 					if v.nm, err = NewLevelDbNeedleMap(v.FileName(".ldb"), indexFile, opts, v.ldbTimeout, v.Version()); err != nil {
 						glog.V(0).Infof("loading leveldb %s error: %v", v.FileName(".ldb"), err)
+						v.nm = nil
+						indexFile.Close()
 					}
 				}
 			}
@@ -351,8 +364,9 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
 		// MaximumNeedleEnd, so this is just a numeric comparison — no extra
 		// disk I/O. A violation marks the volume read-only so a corrupt
 		// .idx left over from a crashed batched write does not silently
-		// power vacuum to drop reachable data. See issue #8928.
-		if !v.HasRemoteFile() && v.nm != nil && v.DataBackend != nil {
+		// power vacuum to drop reachable data. See issue #8928. err == nil
+		// guards against a partial-walk MaximumNeedleEnd.
+		if err == nil && !v.HasRemoteFile() && v.nm != nil && v.DataBackend != nil {
 			if datSize, _, statErr := v.DataBackend.GetStat(); statErr == nil && datSize > 0 {
 				if maxEnd := v.nm.MaxNeedleEnd(); maxEnd > datSize {
 					v.noWriteOrDelete = true

weed/storage/volume_loading_corrupt_idx_test.go

@@ -0,0 +1,41 @@
+package storage
+
+import (
+	"os"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
+// Corrupt .idx made NewSortedFileNeedleMap return a typed-nil into v.nm;
+// the post-load MaxNeedleEnd check then segfaulted on the nil receiver.
+func TestLoad_CorruptIdx_NoSegfault(t *testing.T) {
+	dir := t.TempDir()
+
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("create volume: %v", err)
+	}
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false); err != nil {
+		t.Fatalf("seed write: %v", err)
+	}
+	v.PersistReadOnly(true) // reload goes through SortedFileNeedleMap
+	v.Close()
+
+	// Truncate .idx to a non-aligned size so the walk rejects it.
+	idxPath := VolumeFileName(dir, "", 1) + ".idx"
+	st, err := os.Stat(idxPath)
+	if err != nil {
+		t.Fatalf("stat idx: %v", err)
+	}
+	if err := os.Truncate(idxPath, st.Size()-1); err != nil {
+		t.Fatalf("truncate idx: %v", err)
+	}
+
+	// Pre-fix this panicked inside (*mapMetric).MaxNeedleEnd.
+	v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err == nil {
+		v2.Close()
+	}
+}

weed/topology/allocate_volume.go

@@ -2,6 +2,7 @@ package topology
 
 import (
 	"context"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
@@ -9,6 +10,10 @@ import (
 	"google.golang.org/grpc"
 )
 
+// Cap the RPC so a hung volume server can't strand the layout's
+// growRequest flag and block all future automatic growth.
+const allocateVolumeTimeout = 1 * time.Minute
+
 type AllocateVolumeResult struct {
 	Error string
 }
@@ -17,7 +22,10 @@ func AllocateVolume(dn *DataNode, grpcDialOption grpc.DialOption, vid needle.Vol
 
 	return operation.WithVolumeServerClient(false, dn.ServerAddress(), grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 
-		_, allocateErr := client.AllocateVolume(context.Background(), &volume_server_pb.AllocateVolumeRequest{
+		ctx, cancel := context.WithTimeout(context.Background(), allocateVolumeTimeout)
+		defer cancel()
+
+		_, allocateErr := client.AllocateVolume(ctx, &volume_server_pb.AllocateVolumeRequest{
 			VolumeId:           uint32(vid),
 			Collection:         option.Collection,
 			Replication:        option.ReplicaPlacement.String(),
@@ -36,7 +44,10 @@ func DeleteVolume(dn *DataNode, grpcDialOption grpc.DialOption, vid needle.Volum
 
 	return operation.WithVolumeServerClient(false, dn.ServerAddress(), grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
 
-		_, allocateErr := client.VolumeDelete(context.Background(), &volume_server_pb.VolumeDeleteRequest{
+		ctx, cancel := context.WithTimeout(context.Background(), allocateVolumeTimeout)
+		defer cancel()
+
+		_, allocateErr := client.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{
 			VolumeId: uint32(vid),
 		})
 		return allocateErr

weed/wdclient/filer_client.go

@@ -342,25 +342,37 @@ func (fc *FilerClient) refreshFilerList() {
 		return
 	}
 
-	// Build new filer address list
-	discoveredFilers := make(map[pb.ServerAddress]bool)
+	// Build new filer address set
+	discoveredFilers := make(map[pb.ServerAddress]struct{}, len(updates))
 	for _, update := range updates {
 		if update.Address != "" {
-			discoveredFilers[pb.ServerAddress(update.Address)] = true
+			discoveredFilers[pb.ServerAddress(update.Address)] = struct{}{}
 		}
 	}
 
-	// Thread-safe update of filer list
+	// Ignore snapshots whose addresses are all empty; reconciling against an
+	// empty set would wipe the in-memory list.
+	if len(discoveredFilers) == 0 {
+		glog.V(1).Infof("FilerClient: discovery snapshot for group '%s' had no usable addresses, keeping existing list", fc.filerGroup)
+		return
+	}
+
+	fc.applyDiscoveredFilers(discoveredFilers)
+}
+
+// applyDiscoveredFilers treats the master snapshot as authoritative: survivors
+// keep their health counters, new addresses get fresh health, addresses missing
+// from the snapshot are pruned so replaced pods (e.g. rolled K8s filer pods
+// with new IPs) don't linger and get retried after the circuit-breaker reset.
+func (fc *FilerClient) applyDiscoveredFilers(discoveredFilers map[pb.ServerAddress]struct{}) {
 	fc.filerAddressesMu.Lock()
 	defer fc.filerAddressesMu.Unlock()
 
-	// Build a map of existing filers for efficient O(1) lookup
 	existingFilers := make(map[pb.ServerAddress]struct{}, len(fc.filerAddresses))
 	for _, f := range fc.filerAddresses {
 		existingFilers[f] = struct{}{}
 	}
 
-	// Find new filers - O(N+M) instead of O(N*M)
 	var newFilers []pb.ServerAddress
 	for addr := range discoveredFilers {
 		if _, found := existingFilers[addr]; !found {
@@ -368,20 +380,57 @@ func (fc *FilerClient) refreshFilerList() {
 		}
 	}
 
-	// Add new filers
-	if len(newFilers) > 0 {
-		glog.V(0).Infof("FilerClient: discovered %d new filer(s) in group '%s': %v", len(newFilers), fc.filerGroup, newFilers)
-		fc.filerAddresses = append(fc.filerAddresses, newFilers...)
-
-		// Initialize health tracking for new filers
-		for range newFilers {
-			fc.filerHealth = append(fc.filerHealth, &filerHealth{})
+	var removedFilers []pb.ServerAddress
+	for _, f := range fc.filerAddresses {
+		if _, found := discoveredFilers[f]; !found {
+			removedFilers = append(removedFilers, f)
 		}
 	}
 
-	// Optionally, remove filers that are no longer in the cluster
-	// For now, we keep all filers and rely on health checks to avoid dead ones
-	// This prevents removing filers that might be temporarily unavailable
+	if len(newFilers) == 0 && len(removedFilers) == 0 {
+		return
+	}
+
+	// Remember the active filer so the round-robin pointer can follow it across the rebuild.
+	currentIndex := atomic.LoadInt32(&fc.filerIndex)
+	var currentFiler pb.ServerAddress
+	if currentIndex >= 0 && currentIndex < int32(len(fc.filerAddresses)) {
+		currentFiler = fc.filerAddresses[currentIndex]
+	}
+
+	newAddresses := make([]pb.ServerAddress, 0, len(fc.filerAddresses)-len(removedFilers)+len(newFilers))
+	newHealth := make([]*filerHealth, 0, cap(newAddresses))
+	for i, f := range fc.filerAddresses {
+		if _, found := discoveredFilers[f]; found {
+			newAddresses = append(newAddresses, f)
+			newHealth = append(newHealth, fc.filerHealth[i])
+		}
+	}
+	for _, f := range newFilers {
+		newAddresses = append(newAddresses, f)
+		newHealth = append(newHealth, &filerHealth{})
+	}
+
+	fc.filerAddresses = newAddresses
+	fc.filerHealth = newHealth
+
+	var newIndex int32
+	if currentFiler != "" {
+		for i, f := range newAddresses {
+			if f == currentFiler {
+				newIndex = int32(i)
+				break
+			}
+		}
+	}
+	atomic.StoreInt32(&fc.filerIndex, newIndex)
+
+	if len(removedFilers) > 0 {
+		glog.V(0).Infof("FilerClient: removed %d filer(s) no longer in group '%s': %v", len(removedFilers), fc.filerGroup, removedFilers)
+	}
+	if len(newFilers) > 0 {
+		glog.V(0).Infof("FilerClient: discovered %d new filer(s) in group '%s': %v", len(newFilers), fc.filerGroup, newFilers)
+	}
 }
 
 // GetLookupFileIdFunction returns a lookup function with URL preference handling

weed/wdclient/filer_client_test.go

@@ -0,0 +1,102 @@
+package wdclient
+
+import (
+	"sync/atomic"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+)
+
+func newTestFilerClient(addrs ...pb.ServerAddress) *FilerClient {
+	health := make([]*filerHealth, len(addrs))
+	for i := range health {
+		health[i] = &filerHealth{}
+	}
+	return &FilerClient{
+		filerAddresses: addrs,
+		filerHealth:    health,
+	}
+}
+
+func filerAddressList(fc *FilerClient) []pb.ServerAddress {
+	out := make([]pb.ServerAddress, len(fc.filerAddresses))
+	copy(out, fc.filerAddresses)
+	return out
+}
+
+func TestApplyDiscoveredFilersPrunesStaleAddress(t *testing.T) {
+	a := pb.ServerAddress("10.0.0.1:18888")
+	b := pb.ServerAddress("10.0.0.2:18888") // gets replaced by c
+	c := pb.ServerAddress("10.0.0.3:18888")
+
+	fc := newTestFilerClient(a, b)
+	// Give b a non-trivial failure count so we can confirm it leaves with its health.
+	atomic.StoreInt32(&fc.filerHealth[1].failureCount, 7)
+	// Give a a known failure count so we can confirm survivor health is preserved.
+	atomic.StoreInt32(&fc.filerHealth[0].failureCount, 2)
+	atomic.StoreInt32(&fc.filerIndex, 1) // active filer is b
+
+	fc.applyDiscoveredFilers(map[pb.ServerAddress]struct{}{
+		a: {},
+		c: {},
+	})
+
+	got := filerAddressList(fc)
+	if len(got) != 2 || got[0] != a || got[1] != c {
+		t.Fatalf("expected [%s %s], got %v", a, c, got)
+	}
+	if len(fc.filerHealth) != 2 {
+		t.Fatalf("expected 2 health entries, got %d", len(fc.filerHealth))
+	}
+	if got := atomic.LoadInt32(&fc.filerHealth[0].failureCount); got != 2 {
+		t.Errorf("survivor health was reset: want 2, got %d", got)
+	}
+	if got := atomic.LoadInt32(&fc.filerHealth[1].failureCount); got != 0 {
+		t.Errorf("newly added filer should start with fresh health, got failureCount=%d", got)
+	}
+	// Active filer (b) disappeared; index must reset rather than dangle.
+	if idx := atomic.LoadInt32(&fc.filerIndex); idx != 0 {
+		t.Errorf("expected filerIndex reset to 0 after active filer removed, got %d", idx)
+	}
+}
+
+func TestApplyDiscoveredFilersKeepsIndexOnSurvivor(t *testing.T) {
+	a := pb.ServerAddress("10.0.0.1:18888") // gets replaced
+	b := pb.ServerAddress("10.0.0.2:18888") // active, survives
+	c := pb.ServerAddress("10.0.0.3:18888") // new
+
+	fc := newTestFilerClient(a, b)
+	atomic.StoreInt32(&fc.filerIndex, 1) // active filer is b
+
+	fc.applyDiscoveredFilers(map[pb.ServerAddress]struct{}{
+		b: {},
+		c: {},
+	})
+
+	got := filerAddressList(fc)
+	if len(got) != 2 || got[0] != b || got[1] != c {
+		t.Fatalf("expected [%s %s], got %v", b, c, got)
+	}
+	// b moved to index 0 after a was pruned; index should follow.
+	if idx := atomic.LoadInt32(&fc.filerIndex); idx != 0 {
+		t.Errorf("expected filerIndex to follow surviving active filer to position 0, got %d", idx)
+	}
+}
+
+func TestApplyDiscoveredFilersNoChangeIsNoop(t *testing.T) {
+	a := pb.ServerAddress("10.0.0.1:18888")
+	b := pb.ServerAddress("10.0.0.2:18888")
+
+	fc := newTestFilerClient(a, b)
+	originalHealthA := fc.filerHealth[0]
+	originalHealthB := fc.filerHealth[1]
+
+	fc.applyDiscoveredFilers(map[pb.ServerAddress]struct{}{
+		a: {},
+		b: {},
+	})
+
+	if fc.filerHealth[0] != originalHealthA || fc.filerHealth[1] != originalHealthB {
+		t.Errorf("no-op refresh should not reallocate health entries")
+	}
+}