diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index a6755dd4b..6758976b6 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 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..94278f436 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) diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index 770eb56a9..a0ced7a20 100644 --- a/weed/server/volume_grpc_remote.go +++ b/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. diff --git a/weed/server/volume_grpc_tail.go b/weed/server/volume_grpc_tail.go index 552fea2b7..eaaa3631b 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)