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 b9fcc1d94..65fae7cab 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/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, + ) -> (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(); @@ -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); 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/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 6758976b6..fd8e0deed 100644 --- a/weed/server/volume_grpc_copy.go +++ b/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 diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 94278f436..98a41e755 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/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) } diff --git a/weed/server/volume_grpc_remote.go b/weed/server/volume_grpc_remote.go index a0ced7a20..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" @@ -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. diff --git a/weed/server/volume_grpc_tail.go b/weed/server/volume_grpc_tail.go index eaaa3631b..120675b7f 100644 --- a/weed/server/volume_grpc_tail.go +++ b/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)) } 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) }()