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.

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<tokio::net::TcpStream> {
+    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::<IpAddr>() {
+        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<String> = 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::*;

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};
 

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<Channel, GrpcClientError> {
+    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

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<crate::storage::needle::ttl::TTL>,
+    ) -> (VolumeGrpcService, TempDir) {
+        make_local_service_with_volume_and_trust(collection, ttl, true)
+    }
+
+    fn make_local_service_with_volume_and_trust(
+        collection: &str,
+        ttl: Option<crate::storage::needle::ttl::TTL>,
+        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);

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()

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),

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,

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),

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),

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,
 			}
 		}
 	}

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
+			}
 		}
 	}
 

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...)
 
 }
 

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...)
 }

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

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

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)
+		}
+	}
+}

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)
 	}

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.

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))
 
 }
 

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)
 	}()