volume: validate replica targets and restrict gcs credentials in FetchAndWriteNeedle (#10755)

* volume: validate replica upload targets in FetchAndWriteNeedle

The replica leg forwarded the fetched needle to a caller-supplied address
without checking it, so a malformed target could redirect the upload to an
unintended host or path. Require each replica target to be a bare host:port
whose host is not loopback / link-local / unspecified, reusing the address
deny-list; cluster peers legitimately sit on private networks, so RFC 1918 /
CGNAT stay allowed and -volume.allowUntrustedRemoteEndpoints still opts out.

Validate every target up front so a bad one fails the request before the local
write, and upload through a client that re-checks the resolved address at
connect time so a replica hostname cannot rebind to a blocked address after
validation. Mirrored in Rust (validation moved ahead of the local write; the
Rust S3 path's connect-time re-check is still a follow-up there).

* volume: only accept inline gcs credentials in FetchAndWriteNeedle

The gcs credentials value on this request could name a local filesystem path,
which the SDK reads from disk. Accept only inline JSON here; the server-side
GOOGLE_APPLICATION_CREDENTIALS env var still supplies a path. The Rust volume
server has no gcs backend, so there is nothing to mirror.

seaweed-volume/src/remote_storage/endpoint_guard.rs

@@ -112,6 +112,14 @@ fn embedded_transition_ipv4(v6: Ipv6Addr) -> Option<Ipv4Addr> {
 /// Returns an error if `ip` is not safe to dial from a server that can reach
 /// cluster-internal hosts. Mirrors Go's `checkBlockedIP`.
 pub fn check_blocked_ip(endpoint: &str, ip: IpAddr) -> Result<(), String> {
+    check_blocked_ip_policy(endpoint, ip, false)
+}
+
+/// Like [`check_blocked_ip`], but `allow_private` keeps RFC 1918 / CGNAT
+/// reachable for callers whose target legitimately sits on an internal network
+/// (peer volume servers), while still blocking loopback, link-local (IMDS) and
+/// unspecified. Mirrors Go's `checkBlockedIPPolicy`.
+pub fn check_blocked_ip_policy(endpoint: &str, ip: IpAddr, allow_private: bool) -> Result<(), String> {
     // Normalize IPv4-mapped IPv6 (`::ffff:a.b.c.d`) to its IPv4 form so the
     // IPv4 deny rules apply. The OS routes these to the embedded IPv4 address,
     // so without this `::ffff:127.0.0.1` / `::ffff:169.254.169.254` would slip
@@ -147,17 +155,19 @@ pub fn check_blocked_ip(endpoint: &str, ip: IpAddr) -> Result<(), String> {
             endpoint, ip
         ));
     }
-    if is_private(ip) {
-        return Err(format!(
-            "remote endpoint {:?} resolves to private address {}",
-            endpoint, ip
-        ));
-    }
-    if is_cgnat(ip) {
-        return Err(format!(
-            "remote endpoint {:?} resolves to CGNAT address {}",
-            endpoint, ip
-        ));
+    if !allow_private {
+        if is_private(ip) {
+            return Err(format!(
+                "remote endpoint {:?} resolves to private address {}",
+                endpoint, ip
+            ));
+        }
+        if is_cgnat(ip) {
+            return Err(format!(
+                "remote endpoint {:?} resolves to CGNAT address {}",
+                endpoint, ip
+            ));
+        }
     }
     // IPv6 transition addresses embed an IPv4 destination that routes to the
     // same host wherever the matching relay exists (common in IPv6-only cloud).
@@ -165,7 +175,7 @@ pub fn check_blocked_ip(endpoint: &str, ip: IpAddr) -> Result<(), String> {
     // embedded IPv4 out of the other forms and re-check it against the rules.
     if let IpAddr::V6(v6) = ip {
         if let Some(v4) = embedded_transition_ipv4(v6) {
-            return check_blocked_ip(endpoint, IpAddr::V4(v4));
+            return check_blocked_ip_policy(endpoint, IpAddr::V4(v4), allow_private);
         }
     }
     Ok(())
@@ -287,6 +297,59 @@ pub async fn validate_remote_endpoint(endpoint: &str) -> Result<(), String> {
     }
 }
 
+/// Returns an error if `target` could redirect a replica upload away from a peer
+/// volume server. The target must be a bare `host:port` -- a scheme, userinfo,
+/// path, query or fragment can smuggle a different destination into the
+/// formatted upload URL -- whose host is not loopback, link-local (IMDS) or
+/// unspecified. Cluster peers legitimately sit on private networks, so RFC 1918
+/// / CGNAT are allowed. Mirrors Go's `validateReplicaTarget`.
+pub async fn validate_replica_target(target: &str) -> Result<(), String> {
+    let trimmed = target.trim();
+    if trimmed.is_empty() {
+        return Err("replica target is empty".to_string());
+    }
+    if trimmed.contains("://") || trimmed.contains(['/', '?', '#', '@', '\\']) {
+        return Err(format!("replica target {:?} must be a bare host:port", target));
+    }
+
+    // Require an explicit host:port, handling `[IPv6]:port`. A bracketless IPv6
+    // literal (which carries its own colons) is rejected; peers are addressed as
+    // `[ipv6]:port`, matching Go's net.SplitHostPort.
+    let host = if let Some(rest) = trimmed.strip_prefix('[') {
+        match rest.split_once(']') {
+            Some((h, port)) if port.starts_with(':') && port.len() > 1 => h,
+            _ => return Err(format!("replica target {:?} must be a bare host:port", target)),
+        }
+    } else {
+        match trimmed.rsplit_once(':') {
+            Some((h, port)) if !port.is_empty() && !h.contains(':') => h,
+            _ => return Err(format!("replica target {:?} must be a bare host:port", target)),
+        }
+    };
+
+    if host.is_empty() {
+        return Err(format!("replica target {:?} has no host", target));
+    }
+    if is_blocked_imds_host(&host.to_ascii_lowercase()) {
+        return Err(format!(
+            "replica target {:?} targets instance metadata service",
+            target
+        ));
+    }
+    if let Ok(ip) = host.parse::<IpAddr>() {
+        return check_blocked_ip_policy(target, ip, true);
+    }
+
+    let addrs = resolve_host(host).await?;
+    if addrs.is_empty() {
+        return Err(format!("resolve replica target host {:?}: no addresses", host));
+    }
+    for ip in addrs {
+        check_blocked_ip_policy(target, ip, true)?;
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -443,4 +506,71 @@ mod tests {
             .unwrap_err()
             .contains("loopback"));
     }
+
+    #[test]
+    fn check_blocked_ip_policy_allows_private_peers() {
+        // The replica leg targets peer volume servers, which may be private.
+        assert!(check_blocked_ip_policy("e", ip("10.0.0.7"), true).is_ok());
+        assert!(check_blocked_ip_policy("e", ip("192.168.1.5"), true).is_ok());
+        assert!(check_blocked_ip_policy("e", ip("100.64.0.42"), true).is_ok());
+        // Loopback / IMDS / unspecified stay blocked even when private is allowed.
+        assert!(check_blocked_ip_policy("e", ip("127.0.0.1"), true)
+            .unwrap_err()
+            .contains("loopback"));
+        assert!(check_blocked_ip_policy("e", ip("169.254.169.254"), true)
+            .unwrap_err()
+            .contains("metadata"));
+        assert!(check_blocked_ip_policy("e", ip("0.0.0.0"), true)
+            .unwrap_err()
+            .contains("unspecified"));
+    }
+
+    #[tokio::test]
+    async fn validate_replica_target_rejects_and_allows() {
+        // A path plus a trailing ?a= would otherwise swallow ?type=replicate.
+        assert!(validate_replica_target("127.0.0.1:7000/status/x/?a=")
+            .await
+            .unwrap_err()
+            .contains("bare host:port"));
+        assert!(validate_replica_target("http://10.0.0.7:8080")
+            .await
+            .unwrap_err()
+            .contains("bare host:port"));
+        assert!(validate_replica_target("user@10.0.0.7:8080")
+            .await
+            .unwrap_err()
+            .contains("bare host:port"));
+        assert!(validate_replica_target("10.0.0.7")
+            .await
+            .unwrap_err()
+            .contains("bare host:port"));
+        assert!(validate_replica_target("peer.example.com")
+            .await
+            .unwrap_err()
+            .contains("bare host:port"));
+        assert!(validate_replica_target("127.0.0.1:8080")
+            .await
+            .unwrap_err()
+            .contains("loopback"));
+        assert!(validate_replica_target("[::1]:8080")
+            .await
+            .unwrap_err()
+            .contains("loopback"));
+        assert!(validate_replica_target("169.254.169.254:80")
+            .await
+            .unwrap_err()
+            .contains("metadata"));
+        assert!(validate_replica_target("metadata:80")
+            .await
+            .unwrap_err()
+            .contains("metadata"));
+        assert!(validate_replica_target("")
+            .await
+            .unwrap_err()
+            .contains("empty"));
+        // Legitimate peer volume servers on private networks pass.
+        assert!(validate_replica_target("10.0.0.7:8080").await.is_ok());
+        assert!(validate_replica_target("192.168.1.5:8080").await.is_ok());
+        assert!(validate_replica_target("[fd00::1]:8080").await.is_ok());
+    }
 }

seaweed-volume/src/remote_storage/mod.rs

@@ -7,7 +7,7 @@ pub mod endpoint_guard;
 pub mod s3;
 pub mod s3_tier;
 
-pub use endpoint_guard::validate_remote_endpoint;
+pub use endpoint_guard::{validate_remote_endpoint, validate_replica_target};
 
 use crate::pb::remote_pb::{RemoteConf, RemoteStorageLocation};
 

seaweed-volume/src/server/grpc_server.rs

@@ -3961,6 +3961,21 @@ impl VolumeServer for VolumeGrpcService {
             .as_secs();
         n.set_has_last_modified_date();
 
+        // Validate every replica target before writing anything, so a malformed
+        // or internal target fails the request instead of leaving a local write
+        // behind. Mirrors the Go volume server. NOTE: like the Rust S3 endpoint
+        // guard, this validates the up-front DNS answer but does not yet re-check
+        // at connect time, so a rebinding hostname remains a follow-up.
+        if !self.state.allow_untrusted_remote_endpoints {
+            for replica in &req.replicas {
+                crate::remote_storage::validate_replica_target(&replica.url)
+                    .await
+                    .map_err(|e| {
+                        Status::invalid_argument(format!("reject replica target: {}", e))
+                    })?;
+            }
+        }
+
         // Run local write and replica writes concurrently (matches Go's WaitGroup)
         let mut handles: Vec<tokio::task::JoinHandle<Result<(), String>>> = Vec::new();
 

weed/server/volume_grpc_remote.go

@@ -89,6 +89,15 @@ var imdsIPv4 = net.ParseIP("169.254.169.254")
 var cgnatNet = &net.IPNet{IP: net.IPv4(100, 64, 0, 0), Mask: net.CIDRMask(10, 32)}
 
 func checkBlockedIP(endpoint string, ip net.IP) error {
+	return checkBlockedIPPolicy(endpoint, ip, false)
+}
+
+// checkBlockedIPPolicy rejects addresses that must never be dialed from a
+// server with cluster-internal reach. allowPrivate keeps RFC 1918 / CGNAT
+// reachable for callers whose target legitimately sits on an internal network
+// (peer volume servers), while still blocking loopback, link-local (IMDS) and
+// unspecified.
+func checkBlockedIPPolicy(endpoint string, ip net.IP, allowPrivate bool) error {
 	if ip == nil {
 		return nil
 	}
@@ -104,17 +113,61 @@ func checkBlockedIP(endpoint string, ip net.IP) error {
 		return fmt.Errorf("remote endpoint %q resolves to link-local address %s", endpoint, ip)
 	case ip.IsInterfaceLocalMulticast():
 		return fmt.Errorf("remote endpoint %q resolves to interface-local address %s", endpoint, ip)
-	case ip.IsPrivate():
-		return fmt.Errorf("remote endpoint %q resolves to private address %s", endpoint, ip)
-	case cgnatNet.Contains(ip):
-		return fmt.Errorf("remote endpoint %q resolves to CGNAT address %s", endpoint, ip)
+	}
+	if !allowPrivate {
+		switch {
+		case ip.IsPrivate():
+			return fmt.Errorf("remote endpoint %q resolves to private address %s", endpoint, ip)
+		case cgnatNet.Contains(ip):
+			return fmt.Errorf("remote endpoint %q resolves to CGNAT address %s", endpoint, ip)
+		}
 	}
 	// IPv6 transition addresses embed an IPv4 destination that routes to the
 	// same host wherever the matching relay exists (common in IPv6-only cloud).
 	// net.IP only normalizes ::ffff: mapped addresses, so pull the embedded
 	// IPv4 out of the other forms and re-check it against the deny list.
 	if embedded := embeddedTransitionIPv4(ip); embedded != nil {
-		return checkBlockedIP(endpoint, embedded)
+		return checkBlockedIPPolicy(endpoint, embedded, allowPrivate)
+	}
+	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
+// 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.
+func validateReplicaTarget(ctx context.Context, target string) error {
+	if strings.TrimSpace(target) == "" {
+		return fmt.Errorf("replica target is empty")
+	}
+	if strings.Contains(target, "://") || strings.ContainsAny(target, "/?#@\\") {
+		return fmt.Errorf("replica target %q must be a bare host:port", target)
+	}
+	host, _, splitErr := net.SplitHostPort(target)
+	if splitErr != nil {
+		return fmt.Errorf("replica target %q must be a bare host:port: %w", target, splitErr)
+	}
+	if host == "" {
+		return fmt.Errorf("replica target %q has no host", target)
+	}
+	if _, ok := blockedIMDSHosts[strings.ToLower(host)]; ok {
+		return fmt.Errorf("replica target %q targets instance metadata service", target)
+	}
+	if ip := net.ParseIP(host); ip != nil {
+		return checkBlockedIPPolicy(target, ip, true)
+	}
+	resolveCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+	addrs, lookupErr := lookupIPAddrFunc(resolveCtx, host)
+	if lookupErr != nil {
+		return fmt.Errorf("resolve replica target host %q: %w", host, lookupErr)
+	}
+	for _, addr := range addrs {
+		if err := checkBlockedIPPolicy(target, addr.IP, true); err != nil {
+			return err
+		}
 	}
 	return nil
 }
@@ -160,6 +213,14 @@ func allZero(b []byte) bool {
 // client: even if the attacker's DNS flips to 127.0.0.1 (or any other
 // blocked range) after the up-front check, the dial is refused.
 func guardedDialer(endpoint string) func(ctx context.Context, network, addr string) (net.Conn, error) {
+	return guardedDialerPolicy(endpoint, false)
+}
+
+// guardedDialerPolicy is guardedDialer with the same allowPrivate knob as
+// checkBlockedIPPolicy, so the replica upload path can keep dialing private
+// 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)
@@ -168,7 +229,7 @@ func guardedDialer(endpoint string) func(ctx context.Context, network, addr stri
 		}
 		// If the host is already a literal IP just validate and dial it.
 		if ip := net.ParseIP(host); ip != nil {
-			if err := checkBlockedIP(endpoint, ip); err != nil {
+			if err := checkBlockedIPPolicy(endpoint, ip, allowPrivate); err != nil {
 				return nil, err
 			}
 			return dialer.DialContext(ctx, network, addr)
@@ -183,7 +244,7 @@ func guardedDialer(endpoint string) func(ctx context.Context, network, addr stri
 		}
 		var firstBlockErr error
 		for _, a := range addrs {
-			if err := checkBlockedIP(endpoint, a.IP); err != nil {
+			if err := checkBlockedIPPolicy(endpoint, a.IP, allowPrivate); err != nil {
 				if firstBlockErr == nil {
 					firstBlockErr = err
 				}
@@ -202,6 +263,13 @@ func guardedDialer(endpoint string) func(ctx context.Context, network, addr stri
 // dial addresses that fail checkBlockedIP at connect time. It is meant for
 // per-request use; do not share across remote configs.
 func newGuardedHTTPClient(endpoint string) *http.Client {
+	return newGuardedHTTPClientPolicy(endpoint, false)
+}
+
+// newGuardedHTTPClientPolicy is newGuardedHTTPClient with the allowPrivate knob
+// for the replica upload path, whose targets are cluster peers on private
+// networks.
+func newGuardedHTTPClientPolicy(endpoint string, allowPrivate bool) *http.Client {
 	return &http.Client{
 		Transport: &http.Transport{
 			// No proxy: guardedDialer must see the real target address. Through
@@ -210,7 +278,7 @@ func newGuardedHTTPClient(endpoint string) *http.Client {
 			// dialer exists to close. Operators that need a proxy can opt out
 			// with -volume.allowUntrustedRemoteEndpoints.
 			Proxy:                 nil,
-			DialContext:           guardedDialer(endpoint),
+			DialContext:           guardedDialerPolicy(endpoint, allowPrivate),
 			ForceAttemptHTTP2:     true,
 			MaxIdleConns:          16,
 			IdleConnTimeout:       60 * time.Second,
@@ -242,6 +310,12 @@ func guardedRemoteClient(remoteConf *remote_pb.RemoteConf) (endpoint string, mak
 	return "", nil, false
 }
 
+// gcsCredentialsArePath reports whether a gcs credentials value is a filesystem
+// path rather than inline JSON, matching the gcs client's own inline detection.
+func gcsCredentialsArePath(creds string) bool {
+	return creds != "" && !strings.HasPrefix(creds, "{")
+}
+
 func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_server_pb.FetchAndWriteNeedleRequest) (resp *volume_server_pb.FetchAndWriteNeedleResponse, err error) {
 	if err := vs.checkGrpcAdminAuth(ctx); err != nil {
 		return nil, err
@@ -258,6 +332,15 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 
 	remoteConf := req.RemoteConf
 
+	if !vs.AllowUntrustedRemoteEndpoints && remoteConf != nil {
+		// A gcs credentials value that is a filesystem path is read from disk by
+		// the SDK. Accept only inline JSON on the request; the server env var
+		// still supplies a path.
+		if gcsCredentialsArePath(remoteConf.GetGcsGoogleApplicationCredentials()) {
+			return nil, fmt.Errorf("reject remote credentials: gcs credentials must be inline JSON")
+		}
+	}
+
 	var client remote_storage.RemoteStorageClient
 	var getClientErr error
 	if endpoint, makeClient, ok := guardedRemoteClient(remoteConf); ok && !vs.AllowUntrustedRemoteEndpoints {
@@ -302,6 +385,16 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 		return nil, fmt.Errorf("read from remote %+v: got %d bytes, want %d", remoteStorageLocation, len(data), req.Size)
 	}
 
+	// Validate every replica target before writing anything, so a malformed or
+	// internal target fails the request instead of leaving a local write behind.
+	if !vs.AllowUntrustedRemoteEndpoints {
+		for _, replica := range req.Replicas {
+			if validateErr := validateReplicaTarget(ctx, replica.Url); validateErr != nil {
+				return nil, fmt.Errorf("reject replica target: %w", validateErr)
+			}
+		}
+	}
+
 	var wg sync.WaitGroup
 	var localErr error
 	replicaErrs := make([]error, len(req.Replicas))
@@ -340,10 +433,20 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 					Jwt:               security.EncodedJwt(req.Auth),
 				}
 
-				uploader, uploaderErr := operation.NewUploader()
-				if uploaderErr != nil {
-					replicaErrs[idx] = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, uploaderErr)
-					return
+				// Upload through a client that re-checks the target at connect
+				// time, so a replica hostname cannot rebind to a blocked address
+				// after validateReplicaTarget. Peers may be private, so allow
+				// private here; the opt-out uses the shared global client.
+				var uploader *operation.Uploader
+				if vs.AllowUntrustedRemoteEndpoints {
+					var uploaderErr error
+					uploader, uploaderErr = operation.NewUploader()
+					if uploaderErr != nil {
+						replicaErrs[idx] = fmt.Errorf("remote write needle %d size %d: %v", req.NeedleId, req.Size, uploaderErr)
+						return
+					}
+				} else {
+					uploader = operation.NewUploaderWithHttpClient(newGuardedHTTPClientPolicy(targetVolumeServer, true))
 				}
 
 				if _, replicaWriteErr := uploader.UploadData(ctx, data, uploadOption); replicaWriteErr != nil {

weed/server/volume_grpc_remote_test.go

@@ -374,6 +374,69 @@ func TestRemoteEndpointGuardCoversAzure(t *testing.T) {
 	}
 }
 
+// TestValidateReplicaTarget covers the replica upload leg of
+// FetchAndWriteNeedle. Replica targets are peer volume servers, so unlike the
+// remote endpoint they may sit on a private network; the guard still rejects
+// loopback / link-local / unspecified hosts and any target that is not a bare
+// host:port, since a scheme, path or query would move the upload to a different
+// URL through the format string.
+func TestValidateReplicaTarget(t *testing.T) {
+	originalLookup := lookupIPAddrFunc
+	t.Cleanup(func() { lookupIPAddrFunc = originalLookup })
+
+	lookupIPAddrFunc = stubLookup(t, map[string][]net.IP{
+		"peer.example.com":      {net.ParseIP("10.0.0.7")},
+		"loop.example.com":      {net.ParseIP("127.0.0.1")},
+		"linklocal.example.com": {net.ParseIP("169.254.169.254")},
+	})
+
+	cases := []struct {
+		name    string
+		target  string
+		wantErr bool
+		wantSub string
+	}{
+		// A path plus a trailing "?a=" would otherwise swallow ?type=replicate.
+		{"embedded path and query", "127.0.0.1:7000/status/x/?a=", true, "bare host:port"},
+		{"loopback literal", "127.0.0.1:8080", true, "loopback"},
+		{"ipv6 loopback", "[::1]:8080", true, "loopback"},
+		{"metadata literal", "169.254.169.254:80", true, "metadata"},
+		{"unspecified", "0.0.0.0:8080", true, "unspecified"},
+		{"metadata hostname", "metadata:80", true, "metadata"},
+		{"scheme rejected", "http://10.0.0.7:8080", true, "bare host:port"},
+		{"path rejected", "10.0.0.7:8080/x", true, "bare host:port"},
+		{"query rejected", "10.0.0.7:8080?a=b", true, "bare host:port"},
+		{"userinfo rejected", "user@10.0.0.7:8080", true, "bare host:port"},
+		{"missing port literal", "10.0.0.7", true, "bare host:port"},
+		{"missing port hostname", "peer.example.com", true, "bare host:port"},
+		{"empty", "", true, "empty"},
+		{"resolves to loopback", "loop.example.com:8080", true, "loopback"},
+		{"resolves to link-local", "linklocal.example.com:8080", true, "metadata"},
+		// Legitimate peer volume servers on private networks must pass.
+		{"private peer literal", "10.0.0.7:8080", false, ""},
+		{"private 192 peer", "192.168.1.5:8080", false, ""},
+		{"private peer hostname", "peer.example.com:8080", false, ""},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := validateReplicaTarget(context.Background(), tc.target)
+			if tc.wantErr {
+				if err == nil {
+					t.Fatalf("expected error for %q, got nil", tc.target)
+				}
+				if tc.wantSub != "" && !strings.Contains(err.Error(), tc.wantSub) {
+					t.Fatalf("expected error to contain %q, got %v", tc.wantSub, err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error for %q: %v", tc.target, err)
+			}
+		})
+	}
+}
+
 // TestGuardedRemoteClientSkipsFixedHostBackends confirms backends that only
 // reach a fixed provider host bypass the endpoint guard: azure with no explicit
 // endpoint (public cloud, host derived from the account) and unrelated types.
@@ -412,6 +475,34 @@ func TestGuardedRemoteClientAzureBuildsGuardedClient(t *testing.T) {
 	}
 }
 
+// TestGcsCredentialsArePath confirms a caller-supplied gcs credentials value is
+// only accepted as inline JSON. A filesystem path would otherwise be read from
+// disk by the SDK when handling the request.
+func TestGcsCredentialsArePath(t *testing.T) {
+	paths := []string{
+		"/etc/hostname",
+		"/etc/shadow",
+		"/nope/nothere",
+		"~/creds.json",
+		"relative/creds.json",
+	}
+	for _, p := range paths {
+		if !gcsCredentialsArePath(p) {
+			t.Errorf("expected %q to be treated as a path", p)
+		}
+	}
+	inlineOrEmpty := []string{
+		"",
+		`{"type":"service_account"}`,
+		`{}`,
+	}
+	for _, c := range inlineOrEmpty {
+		if gcsCredentialsArePath(c) {
+			t.Errorf("expected %q to be accepted (inline or empty)", c)
+		}
+	}
+}
+
 // TestGuardedDialerLiteralBlocked confirms that a literal blocked IP target
 // is refused without any DNS lookup.
 func TestGuardedDialerLiteralBlocked(t *testing.T) {
@@ -432,3 +523,46 @@ func TestGuardedDialerLiteralBlocked(t *testing.T) {
 		t.Fatalf("guarded dialer should fail with private-address error, got %v", err)
 	}
 }
+
+// TestGuardedReplicaDialerRebind confirms the replica upload's dial-time guard
+// refuses a hostname that rebinds to loopback after validateReplicaTarget, yet
+// keeps letting private peers through (allowPrivate).
+func TestGuardedReplicaDialerRebind(t *testing.T) {
+	originalLookup := lookupIPAddrFunc
+	t.Cleanup(func() { lookupIPAddrFunc = originalLookup })
+
+	const host = "replica.example.com"
+	var calls atomic.Int32
+	lookupIPAddrFunc = func(_ context.Context, name string) ([]net.IPAddr, error) {
+		if name != host {
+			return nil, &net.DNSError{Err: "no such host", Name: name, IsNotFound: true}
+		}
+		if calls.Add(1) == 1 {
+			return []net.IPAddr{{IP: net.ParseIP("52.216.10.10")}}, nil
+		}
+		return []net.IPAddr{{IP: net.ParseIP("127.0.0.1")}}, nil
+	}
+
+	// Up-front validation sees the public answer and accepts the target.
+	if err := validateReplicaTarget(context.Background(), host+":8080"); err != nil {
+		t.Fatalf("public replica target should validate, got %v", err)
+	}
+	// The dial then re-resolves to loopback and must refuse it.
+	dial := guardedDialerPolicy(host+":8080", true)
+	conn, err := dial(context.Background(), "tcp", host+":8080")
+	if conn != nil {
+		conn.Close()
+		t.Fatalf("guarded replica dialer must refuse loopback rebind, got conn")
+	}
+	if err == nil || !strings.Contains(err.Error(), "loopback") {
+		t.Fatalf("expected loopback refusal, got %v", err)
+	}
+
+	// A private literal peer is allowed through: the dial is attempted (and here
+	// fails on the already-cancelled context) rather than blocked as private.
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, perr := guardedDialerPolicy("10.0.0.5:80", true)(ctx, "tcp", "10.0.0.5:80"); perr != nil && strings.Contains(perr.Error(), "private") {
+		t.Fatalf("private peer must be allowed by the replica dialer, got %v", perr)
+	}
+}