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.

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

@@ -1841,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(
@@ -2925,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(
@@ -3438,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(
@@ -6400,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();
@@ -6472,7 +6492,7 @@ mod tests {
                 crate::remote_storage::s3_tier::S3TierRegistry::new(),
             ),
             read_mode: crate::config::ReadMode::Local,
-            allow_untrusted_remote_endpoints: true,
+            allow_untrusted_remote_endpoints: allow_untrusted,
             master_url: String::new(),
             master_urls: Vec::new(),
             seed_master_set: std::collections::HashSet::new(),
@@ -6496,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);

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/server/volume_grpc_copy.go

@@ -59,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,
@@ -214,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_erasure_coding.go

@@ -392,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 {
@@ -460,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"
@@ -227,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 {
@@ -238,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
@@ -256,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
@@ -265,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

@@ -100,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)
 	}()