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.

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
@@ -2890,6 +2901,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();
@@ -3327,6 +3349,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).
@@ -6439,7 +6472,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: true,
             master_url: String::new(),
             master_urls: Vec::new(),
             seed_master_set: std::collections::HashSet::new(),

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