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.

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

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)
 

weed/server/volume_grpc_remote.go

@@ -137,9 +137,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.

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)