diff --git a/seaweed-volume/src/remote_storage/endpoint_guard.rs b/seaweed-volume/src/remote_storage/endpoint_guard.rs index 82daf88c7..04efe2ea5 100644 --- a/seaweed-volume/src/remote_storage/endpoint_guard.rs +++ b/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 { + 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::() { + 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 = 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::*; diff --git a/seaweed-volume/src/remote_storage/mod.rs b/seaweed-volume/src/remote_storage/mod.rs index 258ba0657..58411bf83 100644 --- a/seaweed-volume/src/remote_storage/mod.rs +++ b/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}; diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs index 4a55992ed..2605f8428 100644 --- a/seaweed-volume/src/server/grpc_client.rs +++ b/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 { + 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 diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index cc312b302..65fae7cab 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/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, + ) -> (VolumeGrpcService, TempDir) { + make_local_service_with_volume_and_trust(collection, ttl, true) + } + + fn make_local_service_with_volume_and_trust( + collection: &str, + ttl: Option, + 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); diff --git a/test/erasure_coding/admin_dockertest/ec_integration_test.go b/test/erasure_coding/admin_dockertest/ec_integration_test.go index 8d62f3279..7840aca1d 100644 --- a/test/erasure_coding/admin_dockertest/ec_integration_test.go +++ b/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() diff --git a/test/erasure_coding/chaos_lifecycle_test.go b/test/erasure_coding/chaos_lifecycle_test.go index 27185f8e3..7d5c9758b 100644 --- a/test/erasure_coding/chaos_lifecycle_test.go +++ b/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), diff --git a/test/erasure_coding/ec_integration_test.go b/test/erasure_coding/ec_integration_test.go index 75aaf5025..f3a146133 100644 --- a/test/erasure_coding/ec_integration_test.go +++ b/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, diff --git a/test/erasure_coding/multidisk_shell_lifecycle_test.go b/test/erasure_coding/multidisk_shell_lifecycle_test.go index 3a9408ddd..58c189910 100644 --- a/test/erasure_coding/multidisk_shell_lifecycle_test.go +++ b/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), diff --git a/test/volume_server/framework/cluster_rust.go b/test/volume_server/framework/cluster_rust.go index b4d188414..88d594d73 100644 --- a/test/volume_server/framework/cluster_rust.go +++ b/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), diff --git a/weed/admin/dash/admin_data.go b/weed/admin/dash/admin_data.go index c57b10a56..f0324a0d7 100644 --- a/weed/admin/dash/admin_data.go +++ b/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, } } } diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index d015d571c..d4485bfcf 100644 --- a/weed/admin/dash/admin_server.go +++ b/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 + } } } diff --git a/weed/operation/grpc_client.go b/weed/operation/grpc_client.go index 9e15ba115..e9063e4b3 100644 --- a/weed/operation/grpc_client.go +++ b/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...) } diff --git a/weed/operation/tail_volume.go b/weed/operation/tail_volume.go index 8decc2df9..5538701ed 100644 --- a/weed/operation/tail_volume.go +++ b/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...) } diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index b508effb4..8ba3a16f3 100644 --- a/weed/pb/grpc_client_server.go +++ b/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 diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index a6755dd4b..fd8e0deed 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 @@ -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 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..98a41e755 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) @@ -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) } diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index 770eb56a9..2627a5ccb 100644 --- a/weed/server/volume_grpc_remote.go +++ b/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. diff --git a/weed/server/volume_grpc_tail.go b/weed/server/volume_grpc_tail.go index 552fea2b7..120675b7f 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) @@ -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)) } diff --git a/weed/shell/command_volume_merge.go b/weed/shell/command_volume_merge.go index 5605ae632..296cdbe8c 100644 --- a/weed/shell/command_volume_merge.go +++ b/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) }()