From a6d72bc272e0619ed20191b6de4b0e624b238c4a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 18 Sep 2026 12:30:07 -0700 Subject: [PATCH 1/2] s3api: delete orphaned chunks only when the entry is confirmed absent (#11389) * s3api: test for chunks deleted under an entry the filer committed Issue #11387: the filer can report a create failure after inserting the entry (e.g. a parent-directory creation failing post-insert). The error arrives in the response rather than as a transport status, so it maps to a definitive error and putToFiler deletes the chunks of the live entry. * s3api: confirmCreateLanded also reports a confirmed-absent entry The verification a failed create runs can answer both directions: the entry matching the uploaded chunks proves the write landed, and an authoritative not-found proves the uploaded chunks are orphaned. Return both outcomes so the cleanup path can gate on the fact rather than the error class. An empty upload can never prove a landing, so a zero-chunk entry match no longer upgrades the outcome. * s3api: delete orphaned chunks only when the entry is confirmed absent A failed create no longer skips verification based on the error class: the filer can fail after inserting the entry (issue #11387) and a partially-applied routed transaction can leave it behind too, both surfacing as definitive errors. Every failed create now resolves the entry's fate, and the uploaded chunks are deleted only when the entry is confirmed absent; anything unverifiable keeps them for vacuum. * s3api: confirm absence on every filer the create could have committed on A lock-path create fails over across filers, so the entry can live on a replica the routed owner has not caught up to; one not-found does not prove absence. The confirmation now queries the owner, the prior owner, and the failover set, declaring absent only when none of them has the entry. * s3api: bound the reconciliation lookups confirmCreateLanded runs The lookups ran on context.Background() under the object write lock, so a connected filer that never replies could stall the write path. One timeout now covers the whole enumeration; an expired budget fails the remaining lookups as uncertain, which keeps the chunks. --- weed/s3api/s3api_object_handlers_put.go | 88 +++++++++++++++---- ...3api_object_handlers_put_ambiguous_test.go | 67 +++++++++++++- weed/s3api/s3api_object_routed_read.go | 8 +- 3 files changed, 136 insertions(+), 27 deletions(-) diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index dd49e442f..5f8510c22 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -22,6 +22,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -995,15 +996,15 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader // If the entry was never created, the uploaded chunks are orphaned and must be deleted. if !entryCreated { - // A transport failure is ambiguous: the filer may have committed the - // entry anyway (issue #11366), so a retryable error never deletes the - // uploaded chunks — it is only upgraded to success when the write owner - // proves the entry landed with these chunks. - ambiguous := createErr != nil && filerErrorToS3Error(createErr) == s3err.ErrServiceUnavailable - if ambiguous && len(chunkResult.FileChunks) > 0 && s3a.confirmCreateLanded(filePath, bucket, object, entry, chunkResult.FileChunks, finalize) { + // A failed create does not prove the entry is absent: a lost response + // can hide a commit (issue #11366) and the filer can fail after + // inserting the entry (issue #11387), so the entry's presence — not + // the error class — decides the chunks' fate. + landed, absent := s3a.confirmCreateLanded(filePath, bucket, object, entry, chunkResult.FileChunks, finalize) + if landed { createCode = s3err.ErrNone } - if createCode != s3err.ErrNone && !ambiguous { + if createCode != s3err.ErrNone && absent { orphaned := chunkResult.FileChunks if manifestChunks, _ := filer.SeparateManifestChunks(entry.GetChunks()); len(manifestChunks) > 0 { orphaned = append(manifestChunks, orphaned...) @@ -1044,28 +1045,51 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader return etag, s3err.ErrNone, responseMetadata } -// confirmCreateLanded checks whether a create that failed ambiguously still -// committed: the stored entry's resolved chunks must be exactly the uploaded -// ones. On a match the finalization the error skipped runs under the object -// write lock, and true reports the write as successful. -func (s3a *S3ApiServer) confirmCreateLanded(filePath, bucket, object string, entry *filer_pb.Entry, uploaded []*filer_pb.FileChunk, finalize *putFinalize) bool { +// createLookupTimeout bounds the entry lookups confirmCreateLanded runs under +// the object write lock, so a hung filer cannot stall the write path. +const createLookupTimeout = 10 * time.Second + +// confirmCreateLanded resolves a create whose outcome is uncertain: a stored +// entry resolving to the uploaded chunks confirms the write landed — the +// finalization the error skipped then runs under the object write lock, and +// landed reports success — while absent requires every filer the create could +// have committed on to lack the entry, the only outcome where the uploaded +// chunks are orphaned. +func (s3a *S3ApiServer) confirmCreateLanded(filePath, bucket, object string, entry *filer_pb.Entry, uploaded []*filer_pb.FileChunk, finalize *putFinalize) (landed, absent bool) { dir, name := path.Dir(filePath), path.Base(filePath) owner := s3a.routableWriteOwner(bucket, object) - confirmed := false + lookupCtx, cancel := context.WithTimeout(context.Background(), createLookupTimeout) + defer cancel() // Verify, finalize, and roll back inside one critical section: a concurrent // write to the same key must not slip in between them. s3a.withObjectWriteLock(bucket, object, nil, func() s3err.ErrorCode { - existing, lookupErr := s3a.lookupEntryPreferringOwner(owner, dir, name) - if lookupErr != nil || existing == nil { + var existing *filer_pb.Entry + uncertain, queried := false, false + for _, target := range s3a.createTargetFilers(owner, bucket, object) { + e, lookupErr := s3a.lookupEntryOnFiler(lookupCtx, target, dir, name) + queried = true + if e != nil { + existing = e + break + } + if lookupErr != nil && !errors.Is(lookupErr, filer_pb.ErrNotFound) { + uncertain = true + } + } + if existing == nil { + absent = queried && !uncertain return s3err.ErrNone } - resolved, _, resolveErr := filer.ResolveChunkManifest(context.Background(), s3a.createLookupFileIdFunction(), existing.GetChunks(), 0, math.MaxInt64, s3a.filerClient) + if len(uploaded) == 0 { + return s3err.ErrNone + } + resolved, _, resolveErr := filer.ResolveChunkManifest(lookupCtx, s3a.createLookupFileIdFunction(), existing.GetChunks(), 0, math.MaxInt64, s3a.filerClient) if resolveErr != nil || !sameFileChunks(resolved, uploaded) { return s3err.ErrNone } glog.Warningf("putToFiler: create entry for %s failed but the entry exists, treating the write as successful", filePath) if finalize == nil || finalize.afterCreate == nil { - confirmed = true + landed = true return s3err.ErrNone } if code := finalize.afterCreate(entry); code != s3err.ErrNone { @@ -1075,10 +1099,36 @@ func (s3a *S3ApiServer) confirmCreateLanded(filePath, bucket, object string, ent } return s3err.ErrNone } - confirmed = true + landed = true return s3err.ErrNone }) - return confirmed + return landed, absent +} + +// createTargetFilers lists the filers a failed create could have committed on: +// the routed owner and the prior one mid-rebalance first, then the failover +// set the lock path dials. Deduped, empty addresses skipped. +func (s3a *S3ApiServer) createTargetFilers(owner pb.ServerAddress, bucket, object string) []pb.ServerAddress { + var filers []pb.ServerAddress + seen := map[pb.ServerAddress]bool{} + add := func(f pb.ServerAddress) { + if f != "" && !seen[f] { + seen[f] = true + filers = append(filers, f) + } + } + add(owner) + add(s3a.priorWriteOwner(bucket, object)) + if s3a.filerClient != nil { + add(s3a.filerClient.GetCurrentFiler()) + for _, f := range s3a.filerClient.GetAllFilers() { + add(f) + } + } + for _, f := range s3a.option.Filers { + add(f) + } + return filers } // sameFileChunks reports whether two chunk lists reference the same needles, diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index 9fd094468..f6ee1e30b 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -88,6 +88,7 @@ type ambiguousPutFiler struct { entries map[string]*filer_pb.Entry apply bool createErr error + respError string lookupErr error lookupFailKey string nextKey uint64 @@ -119,7 +120,7 @@ func (f *ambiguousPutFiler) CreateEntry(_ context.Context, req *filer_pb.CreateE if f.createErr != nil { return nil, f.createErr } - return &filer_pb.CreateEntryResponse{}, nil + return &filer_pb.CreateEntryResponse{Error: f.respError}, nil } func (f *ambiguousPutFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { @@ -148,16 +149,16 @@ func (f *ambiguousPutFiler) LookupVolume(_ context.Context, req *filer_pb.Lookup return resp, nil } -func newPutTestServer(t *testing.T, filerAddr pb.ServerAddress) *S3ApiServer { +func newPutTestServer(t *testing.T, filerAddrs ...pb.ServerAddress) *S3ApiServer { t.Helper() dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) return &S3ApiServer{ option: &S3ApiServerOption{ - Filers: []pb.ServerAddress{filerAddr}, + Filers: filerAddrs, GrpcDialOption: dialOption, BucketsPath: "/buckets", }, - filerClient: wdclient.NewFilerClient([]pb.ServerAddress{filerAddr}, dialOption, ""), + filerClient: wdclient.NewFilerClient(filerAddrs, dialOption, ""), } } @@ -193,6 +194,64 @@ func TestPutToFilerAmbiguousCreateKeepsChunks(t *testing.T) { } } +// Issue 11387: the filer can report a create failure after inserting the entry +// (e.g. a parent-directory creation failing post-insert). The failure arrives +// in the response rather than as a transport status, so it maps to a +// definitive error — but the entry exists and deleting its chunks would +// tombstone live needles. +func TestPutToFilerPostCommitErrorKeepsChunks(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + respError: "create parent directories of /buckets/b: i/o timeout", + } + s3a := newPutTestServer(t, startFakeFiler(t, filerImpl)) + + etag, code := putTestObject(t, s3a) + if code != s3err.ErrNone { + t.Fatalf("putToFiler returned %v, want success once the entry is confirmed on the filer", code) + } + if etag == "" { + t.Fatal("expected an etag") + } + if deleted := volume.deleted(); len(deleted) != 0 { + t.Fatalf("chunks under a live entry were deleted: %v", deleted) + } +} + +// Issue 11387, multi-filer: a create that fails over mid-flight can commit on +// a filer the confirmation does not ask first. A not-found from one replica +// does not authorize deleting chunks an entry on another filer references. +func TestPutToFilerPostCommitErrorOnFailoverFilerKeepsChunks(t *testing.T) { + volume := startFakeVolumeServer(t) + filerA := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: false, + createErr: status.Error(codes.Unavailable, "connect: connection refused"), + } + filerB := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + respError: "create parent directories of /buckets/b: i/o timeout", + } + s3a := newPutTestServer(t, startFakeFiler(t, filerA), startFakeFiler(t, filerB)) + + etag, code := putTestObject(t, s3a) + if code != s3err.ErrNone { + t.Fatalf("putToFiler returned %v, want success once the entry is confirmed on the filer", code) + } + if etag == "" { + t.Fatal("expected an etag") + } + if deleted := volume.deleted(); len(deleted) != 0 { + t.Fatalf("chunks under a live entry were deleted: %v", deleted) + } +} + // A create the filer definitively refused still cleans up the uploaded chunks. func TestPutToFilerConfirmedFailureDeletesOrphans(t *testing.T) { volume := startFakeVolumeServer(t) diff --git a/weed/s3api/s3api_object_routed_read.go b/weed/s3api/s3api_object_routed_read.go index ae8015926..1ff8f7423 100644 --- a/weed/s3api/s3api_object_routed_read.go +++ b/weed/s3api/s3api_object_routed_read.go @@ -54,7 +54,7 @@ func (s3a *S3ApiServer) getObjectEntryRoutedByKey(bucket, object string) (*filer // prior owner once while the ring change is within the cooling-off window. if errors.Is(err, filer_pb.ErrNotFound) { if prior := s3a.priorWriteOwner(bucket, object); prior != "" && prior != owner { - if priorEntry, priorErr := s3a.lookupEntryOnFiler(prior, dir, name); priorErr == nil { + if priorEntry, priorErr := s3a.lookupEntryOnFiler(context.Background(), prior, dir, name); priorErr == nil { return priorEntry, nil } } @@ -87,14 +87,14 @@ func (s3a *S3ApiServer) lookupEntryPreferringOwner(owner pb.ServerAddress, dir, if owner == "" { return s3a.getEntry(dir, name) } - return s3a.lookupEntryOnFiler(owner, dir, name) + return s3a.lookupEntryOnFiler(context.Background(), owner, dir, name) } // lookupEntryOnFiler resolves dir/name against a single filer, without failover. -func (s3a *S3ApiServer) lookupEntryOnFiler(filer pb.ServerAddress, dir, name string) (*filer_pb.Entry, error) { +func (s3a *S3ApiServer) lookupEntryOnFiler(ctx context.Context, filer pb.ServerAddress, dir, name string) (*filer_pb.Entry, error) { var entry *filer_pb.Entry err := pb.WithFilerClient(false, 0, filer, s3a.option.GrpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - resp, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ + resp, lookupErr := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{ Directory: dir, Name: name, }) From 37bf1cd91db6e7997eec3325f455fe06621c9a11 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 18 Sep 2026 12:55:47 -0700 Subject: [PATCH 2/2] volume: validate copy/tail source addresses before dialing (#11390) * pb: stop exiting the process on malformed server addresses ServerToGrpcAddress and GrpcAddressToServerAddress called glog.Fatalf when hostAndPort could not parse the port, which os.Exit(255)ed the whole process. A caller-supplied copy or tail source address reached this path synchronously in the serving goroutine, so one anonymous VolumeCopy with a non-numeric port terminated the volume server. Log the parse error and return the input unchanged instead: the dial or request that consumes the address then fails as an ordinary error. * volume: validate copy and tail source addresses before dialing VolumeCopy, VolumeEcShardsCopy and VolumeTailReceiver dial a caller-supplied source address (SourceDataNode / SourceVolumeServer) with no endpoint validation, so an anonymous caller could aim the volume server at loopback, link-local (cloud metadata) or other unintended destinations and read dial behavior back as a connectivity oracle. Apply the same peer-target deny list FetchAndWriteNeedle uses for replica targets: the source must be a bare host:port whose host is not loopback, link-local or unspecified; cluster peers stay reachable on private networks, and -volume.allowUntrustedRemoteEndpoints opts out. The loopback-using copy tests set the flag to keep exercising the copy path in process. * rust volume: validate copy and tail source addresses before dialing Mirror the Go guard on the Rust volume server: volume_copy, volume_ec_shards_copy and volume_tail_receiver dial a caller-supplied source address, so run it through validate_replica_target first (bare host:port; no loopback, link-local or unspecified hosts; private peers stay allowed). --volume.allowUntrustedRemoteEndpoints opts out; the test fixture and the Rust test-cluster launcher set it so loopback sources in tests keep working. * volume: pin validated copy/tail source addresses at dial time validateReplicaTarget resolves the source hostname once, but the gRPC client resolved it again at connect, leaving a DNS-rebinding window for hostname sources. The copy and tail source dials now run through the same guardedDialerPolicy the remote-storage path uses, so every resolved address is re-checked against the replica deny list (private peers allowed) immediately before the TCP connect. guardedDialerPolicy also moves to util.OutboundDialContext so the guarded path keeps the -ip.bind source binding the default gRPC dialer had. The Rust volume server mirrors this with connect_guarded, a tonic connector that resolves, re-checks each address, and connects to the first passing IP; handlers use it whenever the untrusted-endpoint opt-out is off. A handler-level test now exercises the enabled validation branches for all three source-taking RPCs. * pb: return empty server address for malformed grpc addresses GrpcAddressToServerAddress used to return the unparseable input on a hostAndPort failure, so a malformed raft address (e.g. "host:abc") flowed into admin dashboard master maps unchanged. Return an empty string instead, skip empty conversions at the two raft-cluster merge sites, and drop the now-stale comment about the fatal exit the earlier commit removed. * test: opt erasure-coding loopback clusters out of the remote endpoint guard The erasure-coding suites drive VolumeEcShardsCopy / VolumeCopy between volume servers bound to 127.0.0.1, which the copy/tail source guard now rejects by default. Pass -volume.allowUntrustedRemoteEndpoints to the test volume launches, matching what the volume_server framework harnesses already do. * admin: only claim fallback master leadership on an empty raft response A nonempty RaftListClusterServers response whose entries were all rejected left masterMap empty, so the fallback marked the reachable current master as leader the same way a genuinely empty (non-raft) response does. Track whether the successful response returned zero servers and only promote the fallback master then. --- .../src/remote_storage/endpoint_guard.rs | 50 +++++++ seaweed-volume/src/remote_storage/mod.rs | 4 +- seaweed-volume/src/server/grpc_client.rs | 32 +++++ seaweed-volume/src/server/grpc_server.rs | 131 ++++++++++++++---- .../admin_dockertest/ec_integration_test.go | 2 +- test/erasure_coding/chaos_lifecycle_test.go | 1 + test/erasure_coding/ec_integration_test.go | 6 + .../multidisk_shell_lifecycle_test.go | 1 + test/volume_server/framework/cluster_rust.go | 1 + weed/admin/dash/admin_data.go | 18 ++- weed/admin/dash/admin_server.go | 11 +- weed/operation/grpc_client.go | 9 +- weed/operation/tail_volume.go | 8 +- weed/pb/grpc_client_server.go | 6 +- weed/server/volume_grpc_copy.go | 9 +- weed/server/volume_grpc_copy_verify_test.go | 41 ++++-- weed/server/volume_grpc_erasure_coding.go | 9 +- weed/server/volume_grpc_remote.go | 28 +++- weed/server/volume_grpc_tail.go | 10 +- weed/shell/command_volume_merge.go | 4 +- 20 files changed, 317 insertions(+), 64 deletions(-) diff --git a/seaweed-volume/src/remote_storage/endpoint_guard.rs b/seaweed-volume/src/remote_storage/endpoint_guard.rs index 82daf88c7..04efe2ea5 100644 --- a/seaweed-volume/src/remote_storage/endpoint_guard.rs +++ b/seaweed-volume/src/remote_storage/endpoint_guard.rs @@ -368,6 +368,56 @@ pub async fn validate_replica_target(target: &str) -> Result<(), String> { Ok(()) } +/// Resolve `host`, re-apply the replica deny list (private peers allowed) to +/// every resolved address, and connect to the first one that passes -- the +/// connect-time twin of [`validate_replica_target`], so a hostname whose DNS +/// answer flips to a blocked address after the up-front check is still refused. +/// Mirrors Go's `guardedDialerPolicy` with allowPrivate=true. +pub async fn guarded_tcp_connect( + host: &str, + port: u16, + endpoint: &str, +) -> std::io::Result { + use std::io::{Error, ErrorKind}; + + let denied = |e: String| Error::new(ErrorKind::PermissionDenied, e); + + if is_blocked_imds_host(&host.to_ascii_lowercase()) { + return Err(denied(format!( + "remote endpoint {:?} targets instance metadata service", + endpoint + ))); + } + if let Ok(ip) = host.parse::() { + check_blocked_ip_policy(endpoint, ip, true).map_err(denied)?; + return tokio::net::TcpStream::connect((ip, port)).await; + } + + let lookup = tokio::net::lookup_host((host.to_string(), port)); + let addrs = tokio::time::timeout(std::time::Duration::from_secs(2), lookup) + .await + .map_err(|_| { + Error::new( + ErrorKind::TimedOut, + format!("resolve remote endpoint host {:?}: timed out", host), + ) + })??; + + let mut first_block_err: Option = None; + for addr in addrs { + if let Err(e) = check_blocked_ip_policy(endpoint, addr.ip(), true) { + if first_block_err.is_none() { + first_block_err = Some(e); + } + continue; + } + return tokio::net::TcpStream::connect(addr).await; + } + Err(denied(first_block_err.unwrap_or_else(|| { + format!("resolve remote endpoint host {:?}: no addresses", host) + }))) +} + #[cfg(test)] mod tests { use super::*; diff --git a/seaweed-volume/src/remote_storage/mod.rs b/seaweed-volume/src/remote_storage/mod.rs index 258ba0657..58411bf83 100644 --- a/seaweed-volume/src/remote_storage/mod.rs +++ b/seaweed-volume/src/remote_storage/mod.rs @@ -7,7 +7,9 @@ pub mod endpoint_guard; pub mod s3; pub mod s3_tier; -pub use endpoint_guard::{validate_remote_endpoint, validate_replica_target}; +pub use endpoint_guard::{ + guarded_tcp_connect, validate_remote_endpoint, validate_replica_target, +}; use crate::pb::remote_pb::{RemoteConf, RemoteStorageLocation}; diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index 4a55992ed..2605f8428 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/seaweed-volume/src/server/grpc_client.rs @@ -117,6 +117,38 @@ pub fn build_grpc_endpoint( Ok(endpoint) } +/// Connect `endpoint` through a connector that re-validates every resolved +/// address at connect time (Go's `guardedDialerPolicy` mirror), pinning a +/// validated copy/tail source against DNS rebinding. `allow_untrusted` +/// preserves the plain connect for operators that opted out. +pub async fn connect_guarded( + endpoint: Endpoint, + target: &str, + allow_untrusted: bool, +) -> Result { + if allow_untrusted { + return endpoint + .connect() + .await + .map_err(|e| GrpcClientError(format!("connect {} failed: {}", target, e))); + } + let target_owned = target.to_string(); + let connector = tower::service_fn(move |uri: Uri| { + let target = target_owned.clone(); + async move { + let host = uri.host().unwrap_or_default().to_string(); + let port = uri.port_u16().unwrap_or(80); + crate::remote_storage::guarded_tcp_connect(&host, port, &target) + .await + .map(hyper_util::rt::TokioIo::new) + } + }); + endpoint + .connect_with_connector(connector) + .await + .map_err(|e| GrpcClientError(format!("connect {} failed: {}", target, e))) +} + /// Parse a SeaweedFS server address (`"ip:port.grpcPort"` or /// `"ip:port"`) into the `host:grpcPort` form `build_grpc_endpoint` /// expects. With the trailing `.grpcPort` segment, that segment IS diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index cc312b302..65fae7cab 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1811,6 +1811,17 @@ impl VolumeServer for VolumeGrpcService { let req = request.into_inner(); let vid = VolumeId(req.volume_id); + if !self.state.allow_untrusted_remote_endpoints { + crate::remote_storage::validate_replica_target(&req.source_data_node) + .await + .map_err(|e| { + Status::invalid_argument(format!( + "invalid source data node {}: {}", + req.source_data_node, e + )) + })?; + } + // A pre-existing local replica is NOT deleted up front. Deleting before // the source is confirmed reachable destroys a healthy copy on a // transient source outage (and, on retry, can lose the volume @@ -1830,18 +1841,22 @@ impl VolumeServer for VolumeGrpcService { )) })?; - let channel = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) + let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) .map_err(|e| { Status::internal(format!("VolumeCopy volume {} parse source: {}", vid, e)) - })? - .connect() - .await - .map_err(|e| { - Status::internal(format!( - "VolumeCopy volume {} connect to {}: {}", - vid, grpc_addr, e - )) })?; + let channel = super::grpc_client::connect_guarded( + endpoint, + source, + self.state.allow_untrusted_remote_endpoints, + ) + .await + .map_err(|e| { + Status::internal(format!( + "VolumeCopy volume {} connect to {}: {}", + vid, grpc_addr, e + )) + })?; let mut client = volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( @@ -2890,6 +2905,17 @@ impl VolumeServer for VolumeGrpcService { let req = request.into_inner(); let vid = VolumeId(req.volume_id); + if !self.state.allow_untrusted_remote_endpoints { + crate::remote_storage::validate_replica_target(&req.source_volume_server) + .await + .map_err(|e| { + Status::invalid_argument(format!( + "invalid source volume server {}: {}", + req.source_volume_server, e + )) + })?; + } + // Check volume exists { let store = self.state.store.read().unwrap(); @@ -2903,11 +2929,15 @@ impl VolumeServer for VolumeGrpcService { let grpc_addr = parse_grpc_address(source) .map_err(|e| Status::internal(format!("invalid source address {}: {}", source, e)))?; - let channel = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) - .map_err(|e| Status::internal(format!("parse source: {}", e)))? - .connect() - .await - .map_err(|e| Status::internal(format!("connect to {}: {}", grpc_addr, e)))?; + let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) + .map_err(|e| Status::internal(format!("parse source: {}", e)))?; + let channel = super::grpc_client::connect_guarded( + endpoint, + source, + self.state.allow_untrusted_remote_endpoints, + ) + .await + .map_err(|e| Status::internal(format!("connect to {}: {}", grpc_addr, e)))?; let mut client = volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( @@ -3327,6 +3357,17 @@ impl VolumeServer for VolumeGrpcService { let req = request.into_inner(); let vid = VolumeId(req.volume_id); + if !self.state.allow_untrusted_remote_endpoints { + crate::remote_storage::validate_replica_target(&req.source_data_node) + .await + .map_err(|e| { + Status::invalid_argument(format!( + "invalid source data node {}: {}", + req.source_data_node, e + )) + })?; + } + // Validate wire shard ids at the boundary: ShardId is u8 but only // 0..MAX_SHARD_COUNT are valid. Rejects 256 (would truncate to 0) // and 270 (would alias 14). @@ -3405,21 +3446,25 @@ impl VolumeServer for VolumeGrpcService { )) })?; - let channel = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) + let endpoint = build_grpc_endpoint(&grpc_addr, self.state.outgoing_grpc_tls.as_ref()) .map_err(|e| { Status::internal(format!( "VolumeEcShardsCopy volume {} parse source: {}", vid, e )) - })? - .connect() - .await - .map_err(|e| { - Status::internal(format!( - "VolumeEcShardsCopy volume {} connect to {}: {}", - vid, grpc_addr, e - )) })?; + let channel = super::grpc_client::connect_guarded( + endpoint, + source, + self.state.allow_untrusted_remote_endpoints, + ) + .await + .map_err(|e| { + Status::internal(format!( + "VolumeEcShardsCopy volume {} connect to {}: {}", + vid, grpc_addr, e + )) + })?; let mut client = volume_server_pb::volume_server_client::VolumeServerClient::with_interceptor( @@ -6367,6 +6412,14 @@ mod tests { fn make_local_service_with_volume( collection: &str, ttl: Option, + ) -> (VolumeGrpcService, TempDir) { + make_local_service_with_volume_and_trust(collection, ttl, true) + } + + fn make_local_service_with_volume_and_trust( + collection: &str, + ttl: Option, + allow_untrusted: bool, ) -> (VolumeGrpcService, TempDir) { let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); @@ -6439,7 +6492,7 @@ mod tests { crate::remote_storage::s3_tier::S3TierRegistry::new(), ), read_mode: crate::config::ReadMode::Local, - allow_untrusted_remote_endpoints: false, + allow_untrusted_remote_endpoints: allow_untrusted, master_url: String::new(), master_urls: Vec::new(), seed_master_set: std::collections::HashSet::new(), @@ -6463,6 +6516,36 @@ mod tests { (VolumeGrpcService { state }, tmp) } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_copy_and_tail_handlers_reject_blocked_sources() { + let (service, _tmp) = make_local_service_with_volume_and_trust("guard_rpc", None, false); + + let copy_req = Request::new(volume_server_pb::VolumeCopyRequest { + volume_id: 1, + source_data_node: "169.254.169.254:80".to_string(), + ..Default::default() + }); + let status = service.volume_copy(copy_req).await.err().unwrap(); + assert_eq!(status.code(), tonic::Code::InvalidArgument, "{}", status); + + let tail_req = Request::new(volume_server_pb::VolumeTailReceiverRequest { + volume_id: 1, + source_volume_server: "127.0.0.1:8080".to_string(), + ..Default::default() + }); + let status = service.volume_tail_receiver(tail_req).await.err().unwrap(); + assert_eq!(status.code(), tonic::Code::InvalidArgument, "{}", status); + + let ec_req = Request::new(volume_server_pb::VolumeEcShardsCopyRequest { + volume_id: 1, + source_data_node: "127.0.0.1:8080".to_string(), + shard_ids: vec![0], + ..Default::default() + }); + let status = service.volume_ec_shards_copy(ec_req).await.err().unwrap(); + assert_eq!(status.code(), tonic::Code::InvalidArgument, "{}", status); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_volume_consolidate_index_rpc() { let (service, _tmp) = make_local_service_with_volume("consolidate_rpc", None); diff --git a/test/erasure_coding/admin_dockertest/ec_integration_test.go b/test/erasure_coding/admin_dockertest/ec_integration_test.go index 8d62f3279..7840aca1d 100644 --- a/test/erasure_coding/admin_dockertest/ec_integration_test.go +++ b/test/erasure_coding/admin_dockertest/ec_integration_test.go @@ -129,7 +129,7 @@ func ensureEnvironment(t *testing.T) { port := 8080 + i - 1 dir := filepath.Join("tmp", volName) os.MkdirAll(dir, 0755) - startWeed(t, volName, "volume", "-dir="+dir, "-mserver=localhost:9333", fmt.Sprintf("-port=%d", port), "-ip=localhost") + startWeed(t, volName, "volume", "-dir="+dir, "-mserver=localhost:9333", fmt.Sprintf("-port=%d", port), "-ip=localhost", "-volume.allowUntrustedRemoteEndpoints") }(i) } volWg.Wait() diff --git a/test/erasure_coding/chaos_lifecycle_test.go b/test/erasure_coding/chaos_lifecycle_test.go index 27185f8e3..7d5c9758b 100644 --- a/test/erasure_coding/chaos_lifecycle_test.go +++ b/test/erasure_coding/chaos_lifecycle_test.go @@ -1215,6 +1215,7 @@ func (c *chaosCluster) startVolumeServer(ctx context.Context, i int, logName str "-max", strings.Join(maxVolumes, ","), "-minFreeSpace", "0", "-master", chaosMasterAddr, + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", fmt.Sprintf("rack%d", i), diff --git a/test/erasure_coding/ec_integration_test.go b/test/erasure_coding/ec_integration_test.go index 75aaf5025..f3a146133 100644 --- a/test/erasure_coding/ec_integration_test.go +++ b/test/erasure_coding/ec_integration_test.go @@ -439,6 +439,7 @@ func startSeaweedFSCluster(ctx context.Context, dataDir string) (*TestCluster, e "-dir", volumeDir, "-max", "10", "-master", "127.0.0.1:9333", + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", rack, @@ -1082,6 +1083,7 @@ func startMultiDiskCluster(ctx context.Context, dataDir string) (*MultiDiskClust "-dir", strings.Join(diskDirs, ","), "-max", strings.Join(maxVolumes, ","), "-master", "127.0.0.1:9334", + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", rack, @@ -1478,6 +1480,7 @@ func startClusterWithDiskType(ctx context.Context, dataDir string, diskType stri "-dir", diskDir, "-max", "10", "-mserver", "127.0.0.1:9335", + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", rack, @@ -1708,6 +1711,7 @@ func startMixedDiskTypeCluster(ctx context.Context, dataDir string) (*MultiDiskC "-dir", diskDir, "-max", "10", "-mserver", "127.0.0.1:9336", + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", rack, @@ -2030,6 +2034,7 @@ func startLimitedSsdCluster(ctx context.Context, dataDir string) (*MultiDiskClus "-dir", diskDir, "-max", "10", "-mserver", "127.0.0.1:9337", + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", config.rack, @@ -2112,6 +2117,7 @@ func startMultiRackCluster(ctx context.Context, dataDir string) (*MultiDiskClust "-dir", diskDir, "-max", "10", "-mserver", "127.0.0.1:9338", + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", rack, diff --git a/test/erasure_coding/multidisk_shell_lifecycle_test.go b/test/erasure_coding/multidisk_shell_lifecycle_test.go index 3a9408ddd..58c189910 100644 --- a/test/erasure_coding/multidisk_shell_lifecycle_test.go +++ b/test/erasure_coding/multidisk_shell_lifecycle_test.go @@ -233,6 +233,7 @@ func (c *MultiDiskCluster) startVolumeServers(ctx context.Context) error { "-dir", strings.Join(diskDirs, ","), "-max", strings.Join(maxVolumes, ","), "-master", "127.0.0.1:9334", + "-volume.allowUntrustedRemoteEndpoints", "-ip", "127.0.0.1", "-dataCenter", "dc1", "-rack", fmt.Sprintf("rack%d", i), diff --git a/test/volume_server/framework/cluster_rust.go b/test/volume_server/framework/cluster_rust.go index b4d188414..88d594d73 100644 --- a/test/volume_server/framework/cluster_rust.go +++ b/test/volume_server/framework/cluster_rust.go @@ -201,6 +201,7 @@ func rustVolumeArgs( "--dir", dataDir, "--max", "16", "--master", "127.0.0.1:" + strconv.Itoa(masterPort), + "--volume.allowUntrustedRemoteEndpoints", "--securityFile", filepath.Join(configDir, "security.toml"), "--readMode", profile.ReadMode, "--concurrentUploadLimitMB", strconv.Itoa(profile.ConcurrentUploadLimitMB), diff --git a/weed/admin/dash/admin_data.go b/weed/admin/dash/admin_data.go index c57b10a56..f0324a0d7 100644 --- a/weed/admin/dash/admin_data.go +++ b/weed/admin/dash/admin_data.go @@ -282,7 +282,7 @@ func (s *AdminServer) ShowOverview(w http.ResponseWriter, r *http.Request) { // dashboard never shows an empty list. func (s *AdminServer) getMasterNodesStatus() []MasterNode { masterMap := make(map[string]MasterNode) - raftCallSucceeded := false + raftReturnedEmpty := false err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) @@ -291,16 +291,19 @@ func (s *AdminServer) getMasterNodesStatus() []MasterNode { if err != nil { return err } - raftCallSucceeded = true + raftReturnedEmpty = len(resp.ClusterServers) == 0 for _, server := range resp.ClusterServers { - // pb.GrpcAddressToServerAddress calls glog.Fatalf on a parse - // error, so pre-validate the raft address with net.SplitHostPort - // and skip malformed entries instead of taking the process down. + // Skip malformed raft addresses instead of letting an + // unconvertible value into masterMap. if _, _, splitErr := net.SplitHostPort(server.Address); splitErr != nil { glog.Warningf("skip master with invalid raft address %q: %v", server.Address, splitErr) continue } httpAddress := pb.GrpcAddressToServerAddress(server.Address) + if httpAddress == "" { + glog.Warningf("skip master with invalid raft address %q", server.Address) + continue + } masterMap[httpAddress] = MasterNode{ Address: httpAddress, IsLeader: server.IsLeader, @@ -320,10 +323,11 @@ func (s *AdminServer) getMasterNodesStatus() []MasterNode { addr := pb.ServerAddress(currentMaster).ToHttpAddress() // A successful empty raft response means raft is not initialized // (standalone/non-raft cluster); the only master IS the leader. - // A failed RPC means connectivity issue; do not claim leadership. + // A failed RPC or a nonempty response whose entries were all + // rejected must not claim leadership. masterMap[addr] = MasterNode{ Address: addr, - IsLeader: raftCallSucceeded, + IsLeader: raftReturnedEmpty, } } } diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index d015d571c..d4485bfcf 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -1341,16 +1341,21 @@ func (s *AdminServer) GetClusterMasters() (*ClusterMastersData, error) { } // Then, get additional master information from Raft cluster + raftReturnedEmpty := false err = s.WithMasterClient(func(client master_pb.SeaweedClient) error { resp, err := client.RaftListClusterServers(context.Background(), &master_pb.RaftListClusterServersRequest{}) if err != nil { return err } + raftReturnedEmpty = len(resp.ClusterServers) == 0 // Process each raft server for _, server := range resp.ClusterServers { // Raft stores gRPC addresses, convert to HTTP address httpAddress := pb.GrpcAddressToServerAddress(server.Address) + if httpAddress == "" { + continue + } // Update existing master info or create new one if masterInfo, exists := masterMap[httpAddress]; exists { @@ -1398,10 +1403,12 @@ func (s *AdminServer) GetClusterMasters() (*ClusterMastersData, error) { if currentMaster != "" { masters = append(masters, MasterInfo{ Address: pb.ServerAddress(currentMaster).ToHttpAddress(), - IsLeader: true, + IsLeader: raftReturnedEmpty, Suffrage: "Voter", }) - leaderCount = 1 + if raftReturnedEmpty { + leaderCount = 1 + } } } diff --git a/weed/operation/grpc_client.go b/weed/operation/grpc_client.go index 9e15ba115..e9063e4b3 100644 --- a/weed/operation/grpc_client.go +++ b/weed/operation/grpc_client.go @@ -11,11 +11,18 @@ import ( ) func WithVolumeServerClient(streamingMode bool, volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, fn func(volume_server_pb.VolumeServerClient) error) error { + return WithVolumeServerClientOptions(streamingMode, volumeServer, fn, grpcDialOption) +} + +// WithVolumeServerClientOptions is WithVolumeServerClient with extra dial +// options appended after the TLS option, so a caller dialing an untrusted +// source address can pin the validated endpoint at connect time. +func WithVolumeServerClientOptions(streamingMode bool, volumeServer pb.ServerAddress, fn func(volume_server_pb.VolumeServerClient) error, grpcDialOptions ...grpc.DialOption) error { return pb.WithGrpcClient(context.Background(), streamingMode, 0, func(grpcConnection *grpc.ClientConn) error { client := volume_server_pb.NewVolumeServerClient(grpcConnection) return fn(client) - }, volumeServer.ToGrpcAddress(), false, grpcDialOption) + }, volumeServer.ToGrpcAddress(), false, grpcDialOptions...) } diff --git a/weed/operation/tail_volume.go b/weed/operation/tail_volume.go index 8decc2df9..5538701ed 100644 --- a/weed/operation/tail_volume.go +++ b/weed/operation/tail_volume.go @@ -25,11 +25,11 @@ func TailVolume(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid needle volumeServer := lookup.Locations[0].ServerAddress() - return TailVolumeFromSource(volumeServer, grpcDialOption, vid, sinceNs, timeoutSeconds, fn) + return TailVolumeFromSource(volumeServer, vid, sinceNs, timeoutSeconds, fn, grpcDialOption) } -func TailVolumeFromSource(volumeServer pb.ServerAddress, grpcDialOption grpc.DialOption, vid needle.VolumeId, sinceNs uint64, idleTimeoutSeconds int, fn func(n *needle.Needle) error) error { - return WithVolumeServerClient(true, volumeServer, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { +func TailVolumeFromSource(volumeServer pb.ServerAddress, vid needle.VolumeId, sinceNs uint64, idleTimeoutSeconds int, fn func(n *needle.Needle) error, grpcDialOptions ...grpc.DialOption) error { + return WithVolumeServerClientOptions(true, volumeServer, func(client volume_server_pb.VolumeServerClient) error { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -90,5 +90,5 @@ func TailVolumeFromSource(volumeServer pb.ServerAddress, grpcDialOption grpc.Dia } return nil - }) + }, grpcDialOptions...) } diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index b508effb4..8ba3a16f3 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -551,7 +551,8 @@ func ServerToGrpcAddress(server string) (serverGrpcAddress string) { host, port, parseErr := hostAndPort(server) if parseErr != nil { - glog.Fatalf("server address %s parse error: %v", server, parseErr) + glog.Errorf("server address %s parse error: %v", server, parseErr) + return server } grpcPort := int(port) + 10000 @@ -562,7 +563,8 @@ func ServerToGrpcAddress(server string) (serverGrpcAddress string) { func GrpcAddressToServerAddress(grpcAddress string) (serverAddress string) { host, grpcPort, parseErr := hostAndPort(grpcAddress) if parseErr != nil { - glog.Fatalf("server grpc address %s parse error: %v", grpcAddress, parseErr) + glog.Errorf("server grpc address %s parse error: %v", grpcAddress, parseErr) + return "" } port := int(grpcPort) - 10000 diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index a6755dd4b..fd8e0deed 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -34,6 +34,11 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre if err := vs.CheckMaintenanceMode(); err != nil { return err } + if !vs.AllowUntrustedRemoteEndpoints { + if err := validateReplicaTarget(stream.Context(), req.SourceDataNode); err != nil { + return fmt.Errorf("invalid source data node %s: %w", req.SourceDataNode, err) + } + } // A pre-existing local replica is NOT deleted up front. Deleting before the // source is confirmed reachable destroys a healthy copy on a transient @@ -54,7 +59,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre var sourceVolumeStatusAfterCopy *volume_server_pb.VolumeStatusResponse var dataBaseFileName, indexBaseFileName, idxFileName, datFileName string var hasRemoteDatFile bool - err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + err := operation.WithVolumeServerClientOptions(true, pb.ServerAddress(req.SourceDataNode), func(client volume_server_pb.VolumeServerClient) error { var err error sourceVolumeStatus, err = client.VolumeStatus(stream.Context(), &volume_server_pb.VolumeStatusRequest{ VolumeId: req.VolumeId, @@ -209,7 +214,7 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre } return nil - }) + }, vs.grpcDialOption, vs.guardedGrpcDialOption(req.SourceDataNode)) if err != nil { return err diff --git a/weed/server/volume_grpc_copy_verify_test.go b/weed/server/volume_grpc_copy_verify_test.go index 2c3150f76..415c29d0e 100644 --- a/weed/server/volume_grpc_copy_verify_test.go +++ b/weed/server/volume_grpc_copy_verify_test.go @@ -79,8 +79,9 @@ func runVolumeCopyWithStatusFailure(t *testing.T, failStatusCall int32) (error, targetStore := newVolumeCopyTestStore(t, t.TempDir()) target := &VolumeServer{ - store: targetStore, - grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + AllowUntrustedRemoteEndpoints: true, } err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ VolumeId: uint32(vid), @@ -151,8 +152,9 @@ func TestVolumeCopyKeepsExistingReplicaWhenDestinationFull(t *testing.T) { targetStore.Locations[0].AvailableSpace.Store(0) target := &VolumeServer{ - store: targetStore, - grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + AllowUntrustedRemoteEndpoints: true, } err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ VolumeId: uint32(vid), @@ -191,8 +193,9 @@ func TestVolumeCopyReplacesReplicaAtSlotLimit(t *testing.T) { targetStore.Locations[0].MaxVolumeCount = 1 target := &VolumeServer{ - store: targetStore, - grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: targetStore, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + AllowUntrustedRemoteEndpoints: true, } err := target.VolumeCopy(&volume_server_pb.VolumeCopyRequest{ VolumeId: uint32(vid), @@ -245,8 +248,9 @@ func TestVolumeCopy_KeepsExistingReplicaWhenSourceUnreachable(t *testing.T) { } vs := &VolumeServer{ - store: store, - grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + store: store, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + AllowUntrustedRemoteEndpoints: true, } // 127.0.0.1:1 is unreachable, so ReadVolumeFileStatus on the source fails. @@ -263,3 +267,24 @@ func TestVolumeCopy_KeepsExistingReplicaWhenSourceUnreachable(t *testing.T) { t.Fatalf("existing replica %d was destroyed before the source was verified", vid) } } + +// The copy and tail handlers dial a caller-supplied source address. With the +// default posture (AllowUntrustedRemoteEndpoints unset) a source on a blocked +// address must be rejected before any dial; the opt-out flag restores the old +// behavior for operators whose sources legitimately sit on those ranges. +func TestCopyTailHandlersRejectUntrustedSources(t *testing.T) { + for _, source := range []string{"127.0.0.1:1.10001", "169.254.169.254:0.80"} { + vs := &VolumeServer{ + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + if err := vs.VolumeCopy(&volume_server_pb.VolumeCopyRequest{VolumeId: 1, SourceDataNode: source}, &fakeVolumeCopyStream{}); err == nil { + t.Errorf("VolumeCopy accepted source %q", source) + } + if _, err := vs.VolumeEcShardsCopy(context.Background(), &volume_server_pb.VolumeEcShardsCopyRequest{VolumeId: 1, SourceDataNode: source}); err == nil { + t.Errorf("VolumeEcShardsCopy accepted source %q", source) + } + if _, err := vs.VolumeTailReceiver(context.Background(), &volume_server_pb.VolumeTailReceiverRequest{VolumeId: 1, SourceVolumeServer: source}); err == nil { + t.Errorf("VolumeTailReceiver accepted source %q", source) + } + } +} diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 283f52c9f..98a41e755 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -316,6 +316,11 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv if err := vs.CheckMaintenanceMode(); err != nil { return nil, err } + if !vs.AllowUntrustedRemoteEndpoints { + if err := validateReplicaTarget(ctx, req.SourceDataNode); err != nil { + return nil, fmt.Errorf("invalid source data node %s: %w", req.SourceDataNode, err) + } + } glog.V(0).Infof("VolumeEcShardsCopy: %v", req) @@ -387,7 +392,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv } throttler := util.NewWriteThrottler(ioBytePerSecond) - err := operation.WithVolumeServerClient(true, pb.ServerAddress(req.SourceDataNode), vs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + err := operation.WithVolumeServerClientOptions(true, pb.ServerAddress(req.SourceDataNode), func(client volume_server_pb.VolumeServerClient) error { // copy ec data slices for _, shardId := range req.ShardIds { @@ -455,7 +460,7 @@ func (vs *VolumeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_serv } } return nil - }) + }, vs.grpcDialOption, vs.guardedGrpcDialOption(req.SourceDataNode)) if err != nil { return nil, fmt.Errorf("VolumeEcShardsCopy volume %d: %v", req.VolumeId, err) } diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index 770eb56a9..2627a5ccb 100644 --- a/weed/server/volume_grpc_remote.go +++ b/weed/server/volume_grpc_remote.go @@ -13,6 +13,8 @@ import ( "sync" "time" + "google.golang.org/grpc" + "github.com/seaweedfs/seaweedfs/weed/operation" "github.com/seaweedfs/seaweedfs/weed/pb/remote_pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" @@ -137,9 +139,10 @@ func checkBlockedIPPolicy(endpoint string, ip net.IP, allowPrivate bool) error { return nil } -// validateReplicaTarget rejects a replica upload target that could redirect the -// forwarded write away from a peer volume server. The target must be a bare -// host:port -- a scheme, userinfo, path, query or fragment can smuggle a +// validateReplicaTarget rejects a peer volume server address that could +// redirect a dial away from the cluster: replica upload targets and the +// copy/tail source addresses are all caller-supplied. The target must be a +// bare host:port -- a scheme, userinfo, path, query or fragment can smuggle a // different destination through fmt.Sprintf -- whose host is not loopback, // link-local (IMDS) or unspecified. Cluster peers legitimately sit on private // networks, so RFC 1918 / CGNAT are allowed. @@ -226,7 +229,6 @@ func guardedDialer(endpoint string) func(ctx context.Context, network, addr stri // peers while still refusing loopback / link-local / unspecified at connect // time (closing the rebinding window for replica hostnames too). func guardedDialerPolicy(endpoint string, allowPrivate bool) func(ctx context.Context, network, addr string) (net.Conn, error) { - dialer := &net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second} return func(ctx context.Context, network, addr string) (net.Conn, error) { host, port, splitErr := net.SplitHostPort(addr) if splitErr != nil { @@ -237,7 +239,7 @@ func guardedDialerPolicy(endpoint string, allowPrivate bool) func(ctx context.Co if err := checkBlockedIPPolicy(endpoint, ip, allowPrivate); err != nil { return nil, err } - return dialer.DialContext(ctx, network, addr) + return util.OutboundDialContext(ctx, network, addr) } // Otherwise resolve, validate every answer, and dial the first IP // that passes the deny list. Using a literal-IP target prevents the @@ -255,7 +257,7 @@ func guardedDialerPolicy(endpoint string, allowPrivate bool) func(ctx context.Co } continue } - return dialer.DialContext(ctx, network, net.JoinHostPort(a.IP.String(), port)) + return util.OutboundDialContext(ctx, network, net.JoinHostPort(a.IP.String(), port)) } if firstBlockErr != nil { return nil, firstBlockErr @@ -264,6 +266,20 @@ func guardedDialerPolicy(endpoint string, allowPrivate bool) func(ctx context.Co } } +// guardedGrpcDialOption returns a grpc.DialOption that re-applies the replica +// deny list to every resolved address at connect time, pinning a validated +// copy/tail source against DNS rebinding. It is nil when the operator opted +// out with AllowUntrustedRemoteEndpoints; pb skips nil dial options. +func (vs *VolumeServer) guardedGrpcDialOption(endpoint string) grpc.DialOption { + if vs.AllowUntrustedRemoteEndpoints { + return nil + } + dial := guardedDialerPolicy(endpoint, true) + return grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) { + return dial(ctx, "tcp", addr) + }) +} + // newGuardedHTTPClient returns an *http.Client whose transport refuses to // dial addresses that fail checkBlockedIP at connect time. It is meant for // per-request use; do not share across remote configs. diff --git a/weed/server/volume_grpc_tail.go b/weed/server/volume_grpc_tail.go index 552fea2b7..120675b7f 100644 --- a/weed/server/volume_grpc_tail.go +++ b/weed/server/volume_grpc_tail.go @@ -87,6 +87,12 @@ func (vs *VolumeServer) VolumeTailReceiver(ctx context.Context, req *volume_serv resp := &volume_server_pb.VolumeTailReceiverResponse{} + if !vs.AllowUntrustedRemoteEndpoints { + if err := validateReplicaTarget(ctx, req.SourceVolumeServer); err != nil { + return resp, fmt.Errorf("invalid source volume server %s: %w", req.SourceVolumeServer, err) + } + } + v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) if v == nil { return resp, fmt.Errorf("receiver not found volume id %d", req.VolumeId) @@ -94,10 +100,10 @@ func (vs *VolumeServer) VolumeTailReceiver(ctx context.Context, req *volume_serv defer glog.V(1).Infof("receive tailing volume %d finished", v.Id) - return resp, operation.TailVolumeFromSource(pb.ServerAddress(req.SourceVolumeServer), vs.grpcDialOption, v.Id, req.SinceNs, int(req.IdleTimeoutSeconds), func(n *needle.Needle) error { + return resp, operation.TailVolumeFromSource(pb.ServerAddress(req.SourceVolumeServer), v.Id, req.SinceNs, int(req.IdleTimeoutSeconds), func(n *needle.Needle) error { _, err := vs.store.WriteVolumeNeedle(v.Id, n, false, false) return err - }) + }, vs.grpcDialOption, vs.guardedGrpcDialOption(req.SourceVolumeServer)) } diff --git a/weed/shell/command_volume_merge.go b/weed/shell/command_volume_merge.go index 5605ae632..296cdbe8c 100644 --- a/weed/shell/command_volume_merge.go +++ b/weed/shell/command_volume_merge.go @@ -232,14 +232,14 @@ func startTailNeedleStream(grpcDialOption grpc.DialOption, volumeId needle.Volum ch := make(chan *needle.Needle, 32) stream := &tailNeedleStream{ch: ch} go func() { - err := operation.TailVolumeFromSource(server, grpcDialOption, volumeId, 0, mergeIdleTimeoutSeconds, func(n *needle.Needle) error { + err := operation.TailVolumeFromSource(server, volumeId, 0, mergeIdleTimeoutSeconds, func(n *needle.Needle) error { select { case ch <- n: case <-done: return fmt.Errorf("merge cancelled") } return nil - }) + }, grpcDialOption) close(ch) stream.setErr(err) }()