diff --git a/seaweed-volume/src/main.rs b/seaweed-volume/src/main.rs
index 825ad76056..4e884fe0dd 100644
--- a/seaweed-volume/src/main.rs
+++ b/seaweed-volume/src/main.rs
@@ -217,29 +217,29 @@ fn build_grpc_server_tls_acceptor(
     tls_policy: &TlsPolicy,
     allowed_wildcard_domain: &str,
     allowed_common_names: &[String],
-) -> Option<TlsAcceptor> {
-    if cert_path.is_empty() || key_path.is_empty() || ca_path.is_empty() {
-        return None;
+) -> Result<Option<TlsAcceptor>, Box<dyn std::error::Error>> {
+    if cert_path.is_empty() && key_path.is_empty() {
+        return Ok(None);
+    }
+    if cert_path.is_empty() || key_path.is_empty() {
+        return Err("grpc.volume.cert and grpc.volume.key must both be set".into());
+    }
+    if ca_path.is_empty() {
+        return Err("grpc.volume cert and key are set but grpc.ca is not".into());
     }
     let client_auth_policy = GrpcClientAuthPolicy {
         allowed_common_names: allowed_common_names.to_vec(),
         allowed_wildcard_domain: allowed_wildcard_domain.to_string(),
     };
-    let mut server_config = match build_rustls_server_config_with_grpc_client_auth(
+    let mut server_config = build_rustls_server_config_with_grpc_client_auth(
         cert_path,
         key_path,
         ca_path,
         tls_policy,
         &client_auth_policy,
-    ) {
-        Ok(server_config) => server_config,
-        Err(e) => {
-            warn!("Failed to build gRPC TLS config: {}", e);
-            return None;
-        }
-    };
+    )?;
     server_config.alpn_protocols = vec![b"h2".to_vec()];
-    Some(TlsAcceptor::from(Arc::new(server_config)))
+    Ok(Some(TlsAcceptor::from(Arc::new(server_config))))
 }
 
 fn build_http_server_tls_acceptor(
@@ -540,7 +540,7 @@ async fn run(
         &config.tls_policy,
         &config.grpc_allowed_wildcard_domain,
         &config.grpc_volume_allowed_common_names,
-    );
+    )?;
 
     info!("Starting HTTP server on {}", admin_addr);
     info!("Starting gRPC server on {}", grpc_addr);
@@ -1100,12 +1100,12 @@ mod tests {
 
         assert!(
             build_grpc_server_tls_acceptor(&cert, &key, "", &TlsPolicy::default(), "", &[])
-                .is_none()
+                .is_err()
         );
     }
 
     #[test]
-    fn test_grpc_server_tls_returns_none_when_files_are_missing() {
+    fn test_grpc_server_tls_errors_when_files_are_missing() {
         assert!(
             build_grpc_server_tls_acceptor(
                 "/missing/server.crt",
@@ -1115,12 +1115,12 @@ mod tests {
                 "",
                 &[],
             )
-            .is_none()
+            .is_err()
         );
     }
 
     #[test]
-    fn test_grpc_server_tls_disables_on_unsupported_tls_policy() {
+    fn test_grpc_server_tls_errors_on_unsupported_tls_policy() {
         let dir = tempfile::tempdir().unwrap();
         let cert = write_pem(
             &dir,
@@ -1151,7 +1151,7 @@ mod tests {
                 "",
                 &[],
             )
-            .is_none()
+            .is_err()
         );
     }
 
diff --git a/seaweed-volume/src/security/tls.rs b/seaweed-volume/src/security/tls.rs
index 7dd2094b8b..389603f6d7 100644
--- a/seaweed-volume/src/security/tls.rs
+++ b/seaweed-volume/src/security/tls.rs
@@ -363,8 +363,51 @@ fn common_name_is_allowed(
     allowed_common_names: &HashSet<String>,
     allowed_wildcard_domain: &str,
 ) -> bool {
-    (!allowed_wildcard_domain.is_empty() && common_name.ends_with(allowed_wildcard_domain))
-        || allowed_common_names.contains(common_name)
+    allowed_common_names.contains(common_name)
+        || wildcard_domain_matches(common_name, allowed_wildcard_domain)
+}
+
+// Mirrors the Go wildcard check: the name must end with the domain, carry a
+// non-empty label in front of it, and be a valid DNS name so a CA-signed name
+// like ".example.com" cannot satisfy the suffix check. A leading "*." label
+// is allowed so wildcard certs keep matching. One trailing root dot is
+// ignored on both sides so absolute DNS names still match.
+fn wildcard_domain_matches(name: &str, wildcard_domain: &str) -> bool {
+    let name = name.strip_suffix('.').unwrap_or(name);
+    let wildcard_domain = wildcard_domain.strip_suffix('.').unwrap_or(wildcard_domain);
+    if wildcard_domain.is_empty()
+        || name.len() <= wildcard_domain.len()
+        || !name.ends_with(wildcard_domain)
+    {
+        return false;
+    }
+    if !wildcard_domain.starts_with('.')
+        && name.as_bytes()[name.len() - wildcard_domain.len() - 1] != b'.'
+    {
+        return false;
+    }
+    is_domain_name(name.strip_prefix("*.").unwrap_or(name))
+}
+
+// Mirrors Go's net.IsDomainName: non-empty labels of letters, digits, '-',
+// '_', each up to 63 chars and not '-'-edged, with at least one non-digit so
+// a bare IP cannot pass a domain wildcard.
+fn is_domain_name(name: &str) -> bool {
+    if name.is_empty() || name.len() > 253 {
+        return false;
+    }
+    let mut non_numeric = false;
+    let valid = name.split('.').all(|label| {
+        non_numeric |= label.bytes().any(|c| !c.is_ascii_digit());
+        !label.is_empty()
+            && label.len() <= 63
+            && !label.starts_with('-')
+            && !label.ends_with('-')
+            && label
+                .bytes()
+                .all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
+    });
+    valid && non_numeric
 }
 
 fn go_tls_version_for_supported(version: &SupportedProtocolVersion) -> GoTlsVersion {
@@ -436,10 +479,50 @@ mod tests {
             &allowed_common_names,
             ".example.com",
         ));
+        assert!(common_name_is_allowed(
+            "node.example.com.",
+            &allowed_common_names,
+            ".example.com",
+        ));
+        assert!(common_name_is_allowed(
+            "node.example.com.",
+            &allowed_common_names,
+            ".example.com.",
+        ));
+        assert!(!common_name_is_allowed(
+            "node.example.com..",
+            &allowed_common_names,
+            ".example.com",
+        ));
+        assert!(!common_name_is_allowed(
+            "node..example.com.",
+            &allowed_common_names,
+            ".example.com",
+        ));
         assert!(!common_name_is_allowed(
             "node.prod.other.net",
             &allowed_common_names,
             ".example.com",
         ));
+        assert!(!common_name_is_allowed(
+            ".example.com",
+            &allowed_common_names,
+            ".example.com",
+        ));
+        assert!(!common_name_is_allowed(
+            "node..example.com",
+            &allowed_common_names,
+            ".example.com",
+        ));
+        assert!(!common_name_is_allowed(
+            "badexample.com",
+            &allowed_common_names,
+            "example.com",
+        ));
+        assert!(common_name_is_allowed(
+            "*.example.com",
+            &allowed_common_names,
+            ".example.com",
+        ));
     }
 }
diff --git a/seaweed-volume/src/server/grpc_client.rs b/seaweed-volume/src/server/grpc_client.rs
index 9567e43658..e0772fd326 100644
--- a/seaweed-volume/src/server/grpc_client.rs
+++ b/seaweed-volume/src/server/grpc_client.rs
@@ -60,21 +60,29 @@ pub fn load_outgoing_grpc_tls(
     config: &VolumeServerConfig,
 ) -> Result<Option<OutgoingGrpcTlsConfig>, GrpcClientError> {
     // prefer a dedicated client certificate: CAs may issue certs with only one of the serverAuth/clientAuth EKUs
-    let (cert_file, key_file) = if !config.grpc_client_cert_file.is_empty()
-        && !config.grpc_client_key_file.is_empty()
-    {
+    if config.grpc_client_cert_file.is_empty() != config.grpc_client_key_file.is_empty() {
+        return Err(GrpcClientError(
+            "grpc.volume.client_cert and grpc.volume.client_key must both be set".to_string(),
+        ));
+    }
+    let (cert_file, key_file) = if !config.grpc_client_cert_file.is_empty() {
         (&config.grpc_client_cert_file, &config.grpc_client_key_file)
     } else {
-        if !config.grpc_client_cert_file.is_empty() || !config.grpc_client_key_file.is_empty() {
-            tracing::warn!(
-                "grpc.volume.client_cert and grpc.volume.client_key must both be set, falling back to grpc.volume.cert and grpc.volume.key"
-            );
-        }
         (&config.grpc_cert_file, &config.grpc_key_file)
     };
-    if cert_file.is_empty() || key_file.is_empty() || config.grpc_ca_file.is_empty() {
+    if cert_file.is_empty() && key_file.is_empty() {
         return Ok(None);
     }
+    if cert_file.is_empty() || key_file.is_empty() {
+        return Err(GrpcClientError(
+            "grpc.volume.cert and grpc.volume.key must both be set".to_string(),
+        ));
+    }
+    if config.grpc_ca_file.is_empty() {
+        return Err(GrpcClientError(
+            "grpc.volume cert and key are set but grpc.ca is not".to_string(),
+        ));
+    }
 
     let cert_pem = std::fs::read_to_string(cert_file).map_err(|e| {
         GrpcClientError(format!(
@@ -452,10 +460,21 @@ mod tests {
     }
 
     #[test]
-    fn test_load_outgoing_grpc_tls_requires_cert_key_and_ca() {
+    fn test_load_outgoing_grpc_tls_unconfigured_is_plaintext() {
+        let config = sample_config();
+        assert!(load_outgoing_grpc_tls(&config).unwrap().is_none());
+    }
+
+    #[test]
+    fn test_load_outgoing_grpc_tls_rejects_partial_config() {
         let mut config = sample_config();
         config.grpc_cert_file = "/tmp/client.pem".to_string();
-        assert!(load_outgoing_grpc_tls(&config).unwrap().is_none());
+        assert!(load_outgoing_grpc_tls(&config).is_err());
+
+        let mut config = sample_config();
+        config.grpc_cert_file = "/tmp/client.pem".to_string();
+        config.grpc_key_file = "/tmp/client.key".to_string();
+        assert!(load_outgoing_grpc_tls(&config).is_err());
     }
 
     fn write_pem_files(dir: &tempfile::TempDir, config: &mut VolumeServerConfig) {
@@ -469,6 +488,20 @@ mod tests {
         config.grpc_ca_file = write("ca.pem", "ca");
     }
 
+    #[test]
+    fn test_load_outgoing_grpc_tls_rejects_partial_client_pair() {
+        let dir = tempfile::TempDir::new().unwrap();
+        let mut config = sample_config();
+        write_pem_files(&dir, &mut config);
+        config.grpc_client_cert_file = "/tmp/client.pem".to_string();
+
+        let err = load_outgoing_grpc_tls(&config).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "grpc.volume.client_cert and grpc.volume.client_key must both be set"
+        );
+    }
+
     #[test]
     fn test_load_outgoing_grpc_tls_prefers_client_cert() {
         let dir = tempfile::TempDir::new().unwrap();
diff --git a/weed/command/scaffold/security.toml b/weed/command/scaffold/security.toml
index efa49cce02..f412765b2f 100644
--- a/weed/command/scaffold/security.toml
+++ b/weed/command/scaffold/security.toml
@@ -83,10 +83,23 @@ expires_after_seconds = 10           # seconds
 # the component reuses cert/key for both directions.
 # If client and server certificates come from different issuing CAs, put both
 # CA certificates in the ca PEM file.
+# Once a component's cert/key are set, any error loading them or grpc.ca is
+# fatal: the component never falls back to plaintext.
 [grpc]
 ca = ""
 # Set wildcard domain for enable TLS authentication by common names
 allowed_wildcard_domain = "" # .mycompany.com
+# Optional allowlist a client checks the dialed server's certificate against:
+# the subject common name, a DNS SAN, or a URI SAN (e.g. a SPIFFE id) must be
+# listed, or match allowed_server_wildcard_domain, else the handshake fails.
+# With a per-workload CA, list the identities each server may present so one
+# workload cannot impersonate another component. Unset accepts any CA-signed
+# server certificate, as before. Any [grpc.<component>] section may set its own
+# allowed_server_commonNames to override this list, and scope it further per
+# dialed target, e.g. the filer reads [grpc.filer]
+# volume_allowed_server_commonNames for volume server calls.
+allowed_server_commonNames = ""      # comma-separated server identities
+allowed_server_wildcard_domain = ""  # .mycompany.com
 
 # Volume server gRPC options (server-side)
 # Enables mTLS for incoming gRPC connections to volume server
diff --git a/weed/filer/filer.go b/weed/filer/filer.go
index 06168a8006..bce66f5889 100644
--- a/weed/filer/filer.go
+++ b/weed/filer/filer.go
@@ -46,6 +46,14 @@ var (
 	OS_GID = uint32(os.Getgid())
 )
 
+// DialOptions selects the outbound dial option per target kind. Empty fields
+// fall back to the filer option, so sharing one option keeps past behavior.
+type DialOptions struct {
+	Filer  grpc.DialOption
+	Master grpc.DialOption
+	Volume grpc.DialOption
+}
+
 type Filer struct {
 	UniqueFilerId       int32
 	UniqueFilerEpoch    int32
@@ -53,6 +61,7 @@ type Filer struct {
 	MasterClient        *wdclient.MasterClient
 	FileIdDeletionQueue *util.UnboundedQueue
 	GrpcDialOption      grpc.DialOption
+	dialOptions         *DialOptions
 	DirBucketsPath      string
 	Cipher              bool
 	LocalMetaLogBuffer  *log_buffer.LogBuffer
@@ -123,11 +132,12 @@ type Filer struct {
 	flushCancel context.CancelFunc
 }
 
-func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerHost pb.ServerAddress, filerGroup string, collection string, replication string, dataCenter string, maxFilenameLength uint32, notifyFn func()) *Filer {
+func NewFiler(masters pb.ServerDiscovery, dialOptions *DialOptions, filerHost pb.ServerAddress, filerGroup string, collection string, replication string, dataCenter string, maxFilenameLength uint32, notifyFn func()) *Filer {
 	f := &Filer{
-		MasterClient:        wdclient.NewMasterClient(grpcDialOption, filerGroup, cluster.FilerType, filerHost, dataCenter, "", masters),
+		MasterClient:        wdclient.NewMasterClient(dialOptions.master(), filerGroup, cluster.FilerType, filerHost, dataCenter, "", masters),
 		FileIdDeletionQueue: util.NewUnboundedQueue(),
-		GrpcDialOption:      grpcDialOption,
+		GrpcDialOption:      dialOptions.filer(),
+		dialOptions:         dialOptions,
 		FilerConf:           NewFilerConf(),
 		RemoteStorage:       NewFilerRemoteStorage(),
 		UniqueFilerId:       util.RandomInt32(),
@@ -177,6 +187,47 @@ func NewFiler(masters pb.ServerDiscovery, grpcDialOption grpc.DialOption, filerH
 	return f
 }
 
+func (o *DialOptions) filer() grpc.DialOption {
+	if o == nil {
+		return nil
+	}
+	return o.Filer
+}
+
+func (o *DialOptions) master() grpc.DialOption {
+	if o == nil {
+		return nil
+	}
+	if o.Master != nil {
+		return o.Master
+	}
+	return o.Filer
+}
+
+func (o *DialOptions) volume() grpc.DialOption {
+	if o == nil {
+		return nil
+	}
+	if o.Volume != nil {
+		return o.Volume
+	}
+	return o.Filer
+}
+
+func (f *Filer) masterGrpcDialOption() grpc.DialOption {
+	if f.dialOptions != nil && f.dialOptions.Master != nil {
+		return f.dialOptions.Master
+	}
+	return f.GrpcDialOption
+}
+
+func (f *Filer) volumeGrpcDialOption() grpc.DialOption {
+	if f.dialOptions != nil && f.dialOptions.Volume != nil {
+		return f.dialOptions.Volume
+	}
+	return f.GrpcDialOption
+}
+
 func (f *Filer) buildRemoteStorageClient(ctx context.Context, remoteConf *remote_pb.RemoteConf) (remote_storage.RemoteStorageClient, error) {
 	if f.BuildGuardedRemoteClient != nil {
 		return f.BuildGuardedRemoteClient(ctx, remoteConf, f.AllowUntrustedRemoteEndpoints)
@@ -294,7 +345,7 @@ func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*m
 			break
 		}
 		lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout)
-		l, lookErr := cluster.LookupClusterLeader(lookupCtx, master, f.GrpcDialOption)
+		l, lookErr := cluster.LookupClusterLeader(lookupCtx, master, f.masterGrpcDialOption())
 		cancel()
 		if lookErr != nil {
 			queryErrs = append(queryErrs, lookErr)
@@ -309,7 +360,7 @@ func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*m
 		settleDeadline := time.Now().Add(leaderSettleTimeout)
 		for {
 			lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout)
-			nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, leader, f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType)
+			nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, leader, f.masterGrpcDialOption(), f.MasterClient.FilerGroup, cluster.FilerType)
 			cancel()
 			if listErr != nil {
 				queryErrs = append(queryErrs, fmt.Errorf("list peers from leader %s: %w", leader, listErr))
@@ -334,7 +385,7 @@ func (f *Filer) ListExistingPeerUpdates(ctx context.Context) (existingNodes []*m
 			break
 		}
 		lookupCtx, cancel := context.WithTimeout(ctx, peerLookupTimeout)
-		nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, master, f.GrpcDialOption, f.MasterClient.FilerGroup, cluster.FilerType)
+		nodes, listErr := cluster.ListExistingPeerUpdates(lookupCtx, master, f.masterGrpcDialOption(), f.MasterClient.FilerGroup, cluster.FilerType)
 		cancel()
 		if listErr != nil {
 			queryErrs = append(queryErrs, listErr)
diff --git a/weed/filer/filer_deletion.go b/weed/filer/filer_deletion.go
index c2944b82bf..0fe0f62d96 100644
--- a/weed/filer/filer_deletion.go
+++ b/weed/filer/filer_deletion.go
@@ -362,7 +362,7 @@ func (f *Filer) processDeletionBatch(ctx context.Context, toDeleteFileIds []stri
 	}
 
 	// Delete files and classify outcomes
-	outcomes := deleteFilesAndClassify(ctx, f.GrpcDialOption, uniqueFileIdsSlice, lookupFunc)
+	outcomes := deleteFilesAndClassify(ctx, f.volumeGrpcDialOption(), uniqueFileIdsSlice, lookupFunc)
 
 	// Process outcomes
 	var successCount, notFoundCount, retryableErrorCount, permanentErrorCount int
@@ -582,7 +582,7 @@ func (f *Filer) processRetryBatch(readyItems []*DeletionRetryItem, lookupFunc fu
 	}
 
 	// Delete files and classify outcomes
-	outcomes := deleteFilesAndClassify(context.Background(), f.GrpcDialOption, fileIds, lookupFunc)
+	outcomes := deleteFilesAndClassify(context.Background(), f.volumeGrpcDialOption(), fileIds, lookupFunc)
 
 	// Process outcomes - iterate over readyItems to ensure all items are accounted for
 	var successCount, notFoundCount, retryCount, permanentErrorCount int
diff --git a/weed/filer/filer_notify_append.go b/weed/filer/filer_notify_append.go
index 24f4e733b9..27bda98fdc 100644
--- a/weed/filer/filer_notify_append.go
+++ b/weed/filer/filer_notify_append.go
@@ -86,7 +86,7 @@ func (f *Filer) assignAndUpload(ctx context.Context, targetFile string, data []b
 		ExpectedDataSize:    uint64(len(data)),
 	}
 
-	assignResult, err := operation.Assign(ctx, f.GetMaster, f.GrpcDialOption, assignRequest)
+	assignResult, err := operation.Assign(ctx, f.GetMaster, f.masterGrpcDialOption(), assignRequest)
 	if err != nil {
 		return nil, nil, fmt.Errorf("AssignVolume: %w", err)
 	}
diff --git a/weed/security/tls.go b/weed/security/tls.go
index 4bdf9e89e2..88417db533 100644
--- a/weed/security/tls.go
+++ b/weed/security/tls.go
@@ -67,8 +67,16 @@ func (s *SNIStrippingTransportCredentials) OverrideServerName(serverNameOverride
 }
 
 func LoadServerTLS(config *util.ViperProxy, component string) (grpc.ServerOption, grpc.ServerOption) {
+	creds, verify, err := loadServerTLS(config, component)
+	if err != nil {
+		glog.Fatalf("gRPC server TLS for %s: %v", component, err)
+	}
+	return creds, verify
+}
+
+func loadServerTLS(config *util.ViperProxy, component string) (grpc.ServerOption, grpc.ServerOption, error) {
 	if config == nil {
-		return nil, nil
+		return nil, nil, nil
 	}
 
 	serverOptions := pemfile.Options{
@@ -76,14 +84,19 @@ func LoadServerTLS(config *util.ViperProxy, component string) (grpc.ServerOption
 		KeyFile:         config.GetString(component + ".key"),
 		RefreshDuration: CredRefreshingInterval,
 	}
+	if serverOptions.CertFile == "" && serverOptions.KeyFile == "" {
+		return nil, nil, nil
+	}
 	if serverOptions.CertFile == "" || serverOptions.KeyFile == "" {
-		return nil, nil
+		return nil, nil, fmt.Errorf("%s.cert and %s.key must both be set", component, component)
+	}
+	if err := checkTLSFiles(serverOptions.CertFile, serverOptions.KeyFile, config.GetString("grpc.ca")); err != nil {
+		return nil, nil, err
 	}
 
 	serverIdentityProvider, err := pemfile.NewProvider(serverOptions)
 	if err != nil {
-		glog.Warningf("pemfile.NewProvider(%v) %v failed: %v", serverOptions, component, err)
-		return nil, nil
+		return nil, nil, fmt.Errorf("pemfile.NewProvider(%v) failed: %w", serverOptions, err)
 	}
 
 	serverRootOptions := pemfile.Options{
@@ -92,8 +105,7 @@ func LoadServerTLS(config *util.ViperProxy, component string) (grpc.ServerOption
 	}
 	serverRootProvider, err := pemfile.NewProvider(serverRootOptions)
 	if err != nil {
-		glog.Warningf("pemfile.NewProvider(%v) failed: %v", serverRootOptions, err)
-		return nil, nil
+		return nil, nil, fmt.Errorf("pemfile.NewProvider(%v) failed: %w", serverRootOptions, err)
 	}
 
 	// Start a server and create a client using advancedtls API with Provider.
@@ -109,30 +121,21 @@ func LoadServerTLS(config *util.ViperProxy, component string) (grpc.ServerOption
 	}
 	options.MinTLSVersion, err = TlsVersionByName(config.GetString("tls.min_version"))
 	if err != nil {
-		glog.Warningf("tls min version parse failed, %v", err)
-		return nil, nil
+		return nil, nil, err
 	}
 	options.MaxTLSVersion, err = TlsVersionByName(config.GetString("tls.max_version"))
 	if err != nil {
-		glog.Warningf("tls max version parse failed, %v", err)
-		return nil, nil
+		return nil, nil, err
 	}
 	options.CipherSuites, err = TlsCipherSuiteByNames(config.GetString("tls.cipher_suites"))
 	if err != nil {
-		glog.Warningf("tls cipher suite parse failed, %v", err)
-		return nil, nil
+		return nil, nil, err
 	}
-	allowedCommonNames := config.GetString(component + ".allowed_commonNames")
-	allowedWildcardDomain := config.GetString("grpc.allowed_wildcard_domain")
-	if allowedCommonNames != "" || allowedWildcardDomain != "" {
-		allowedCommonNamesMap := make(map[string]bool)
-		for _, s := range strings.Split(allowedCommonNames, ",") {
-			allowedCommonNamesMap[s] = true
-		}
-		auther := Authenticator{
-			AllowedCommonNames:    allowedCommonNamesMap,
-			AllowedWildcardDomain: allowedWildcardDomain,
-		}
+	auther := Authenticator{
+		AllowedCommonNames:    parseAllowedCommonNames(config.GetString(component + ".allowed_commonNames")),
+		AllowedWildcardDomain: config.GetString("grpc.allowed_wildcard_domain"),
+	}
+	if len(auther.AllowedCommonNames) != 0 || auther.AllowedWildcardDomain != "" {
 		options.AdditionalPeerVerification = auther.Authenticate
 	} else {
 		options.AdditionalPeerVerification = func(params *advancedtls.HandshakeVerificationInfo) (*advancedtls.PostHandshakeVerificationResults, error) {
@@ -141,10 +144,9 @@ func LoadServerTLS(config *util.ViperProxy, component string) (grpc.ServerOption
 	}
 	ta, err := advancedtls.NewServerCreds(options)
 	if err != nil {
-		glog.Warningf("advancedtls.NewServerCreds(%v) failed: %v", options, err)
-		return nil, nil
+		return nil, nil, fmt.Errorf("advancedtls.NewServerCreds failed: %w", err)
 	}
-	return grpc.Creds(ta), nil
+	return grpc.Creds(ta), nil, nil
 }
 
 func LoadClientTLSFromFile(configFile string, component string) (grpc.DialOption, error) {
@@ -165,21 +167,56 @@ func LoadClientTLSFromFile(configFile string, component string) (grpc.DialOption
 }
 
 func LoadClientTLS(config *util.ViperProxy, component string) grpc.DialOption {
+	option, err := loadClientTLS(config, component, component)
+	if err != nil {
+		glog.Fatalf("gRPC client TLS for %s: %v", component, err)
+	}
+	return option
+}
+
+// LoadClientTLSForTarget builds the component's client dial option while
+// selecting the server-identity allowlist for a specific target kind: a filer
+// dialing a volume keeps presenting its grpc.filer client certificate but
+// verifies the peer against grpc.filer.volume_allowed_server_commonNames. The
+// server-name and wildcard keys fall back from the client's per-target key to
+// the component key to grpc.*, so a deployment without per-target keys
+// verifies exactly as before. The per-target key lives under the client
+// component because grpc.<component>.allowed_server_commonNames denotes that
+// component's own outbound allowlist.
+func LoadClientTLSForTarget(config *util.ViperProxy, clientComponent, targetComponent string) grpc.DialOption {
+	option, err := loadClientTLS(config, clientComponent, targetComponent)
+	if err != nil {
+		glog.Fatalf("gRPC client TLS for %s: %v", clientComponent, err)
+	}
+	return option
+}
+
+func loadClientTLS(config *util.ViperProxy, component, targetComponent string) (grpc.DialOption, error) {
 	if config == nil {
-		return grpc.WithTransportCredentials(insecure.NewCredentials())
+		return grpc.WithTransportCredentials(insecure.NewCredentials()), nil
 	}
 
 	// prefer a dedicated client certificate: CAs may issue certs with only one of the serverAuth/clientAuth EKUs
 	certFileName, keyFileName := config.GetString(component+".client_cert"), config.GetString(component+".client_key")
+	partialClientPair := (certFileName == "") != (keyFileName == "")
+	if partialClientPair {
+		return nil, fmt.Errorf("%s.client_cert and %s.client_key must both be set", component, component)
+	}
 	if certFileName == "" || keyFileName == "" {
-		if certFileName != "" || keyFileName != "" {
-			glog.Warningf("%s.client_cert and %s.client_key must both be set, falling back to %s.cert and %s.key", component, component, component, component)
-		}
 		certFileName, keyFileName = config.GetString(component+".cert"), config.GetString(component+".key")
 	}
+	if certFileName == "" && keyFileName == "" {
+		return grpc.WithTransportCredentials(insecure.NewCredentials()), nil
+	}
+	if certFileName == "" || keyFileName == "" {
+		return nil, fmt.Errorf("%s.cert and %s.key must both be set", component, component)
+	}
 	caFileName := config.GetString("grpc.ca")
-	if certFileName == "" || keyFileName == "" || caFileName == "" {
-		return grpc.WithTransportCredentials(insecure.NewCredentials())
+	if caFileName == "" {
+		return nil, fmt.Errorf("%s cert and key are set but grpc.ca is not", component)
+	}
+	if err := checkTLSFiles(certFileName, keyFileName, caFileName); err != nil {
+		return nil, err
 	}
 
 	clientOptions := pemfile.Options{
@@ -189,37 +226,54 @@ func LoadClientTLS(config *util.ViperProxy, component string) grpc.DialOption {
 	}
 	clientProvider, err := pemfile.NewProvider(clientOptions)
 	if err != nil {
-		glog.Warningf("pemfile.NewProvider(%v) failed %v", clientOptions, err)
-		return grpc.WithTransportCredentials(insecure.NewCredentials())
+		return nil, fmt.Errorf("pemfile.NewProvider(%v) failed: %w", clientOptions, err)
 	}
 	clientRootOptions := pemfile.Options{
-		RootFile:        config.GetString("grpc.ca"),
+		RootFile:        caFileName,
 		RefreshDuration: CredRefreshingInterval,
 	}
 	clientRootProvider, err := pemfile.NewProvider(clientRootOptions)
 	if err != nil {
-		glog.Warningf("pemfile.NewProvider(%v) failed: %v", clientRootOptions, err)
-		return grpc.WithTransportCredentials(insecure.NewCredentials())
+		return nil, fmt.Errorf("pemfile.NewProvider(%v) failed: %w", clientRootOptions, err)
 	}
 	options := &advancedtls.Options{
 		IdentityOptions: advancedtls.IdentityCertificateOptions{
 			IdentityProvider: clientProvider,
 		},
-		AdditionalPeerVerification: func(params *advancedtls.HandshakeVerificationInfo) (*advancedtls.PostHandshakeVerificationResults, error) {
-			return &advancedtls.PostHandshakeVerificationResults{}, nil
-		},
 		RootOptions: advancedtls.RootCertificateOptions{
 			RootProvider: clientRootProvider,
 		},
 		VerificationType: advancedtls.CertVerification,
 	}
+	targetPrefix := ""
+	if targetComponent != component {
+		targetPrefix = strings.TrimPrefix(targetComponent, "grpc.") + "_"
+	}
+	allowedServerNames := firstConfigString(config,
+		component+"."+targetPrefix+"allowed_server_commonNames",
+		component+".allowed_server_commonNames",
+		"grpc.allowed_server_commonNames")
+	allowedWildcardDomain := firstConfigString(config,
+		component+"."+targetPrefix+"allowed_server_wildcard_domain",
+		component+".allowed_server_wildcard_domain",
+		"grpc.allowed_server_wildcard_domain")
+	auther := Authenticator{
+		AllowedCommonNames:    parseAllowedCommonNames(allowedServerNames),
+		AllowedWildcardDomain: allowedWildcardDomain,
+	}
+	if len(auther.AllowedCommonNames) != 0 || auther.AllowedWildcardDomain != "" {
+		options.AdditionalPeerVerification = auther.AuthenticateServer
+	} else {
+		options.AdditionalPeerVerification = func(params *advancedtls.HandshakeVerificationInfo) (*advancedtls.PostHandshakeVerificationResults, error) {
+			return &advancedtls.PostHandshakeVerificationResults{}, nil
+		}
+	}
 	ta, err := advancedtls.NewClientCreds(options)
 	if err != nil {
-		glog.Warningf("advancedtls.NewClientCreds(%v) failed: %v", options, err)
-		return grpc.WithTransportCredentials(insecure.NewCredentials())
+		return nil, fmt.Errorf("advancedtls.NewClientCreds failed: %w", err)
 	}
 	wrapped := &SNIStrippingTransportCredentials{creds: ta}
-	return grpc.WithTransportCredentials(wrapped)
+	return grpc.WithTransportCredentials(wrapped), nil
 }
 
 // LoadHTTPClientFromFile creates an HTTP client using the https.client TLS
@@ -272,8 +326,88 @@ func LoadClientTLSHTTP(clientCertFile string) *tls.Config {
 	}
 }
 
+// checkTLSFiles validates configured TLS material up front: the pemfile
+// providers read lazily, so without this a typo'd or unparseable file would
+// only surface at handshake time instead of at startup.
+func checkTLSFiles(certFile, keyFile, caFile string) error {
+	if _, err := tls.LoadX509KeyPair(certFile, keyFile); err != nil {
+		return err
+	}
+	if caFile == "" {
+		return nil
+	}
+	caPem, err := os.ReadFile(caFile)
+	if err != nil {
+		return err
+	}
+	if !x509.NewCertPool().AppendCertsFromPEM(caPem) {
+		return fmt.Errorf("no CA certificates found in %s", caFile)
+	}
+	return nil
+}
+
+func parseAllowedCommonNames(commonNames string) map[string]bool {
+	allowed := make(map[string]bool)
+	for _, s := range strings.Split(commonNames, ",") {
+		if s = strings.TrimSpace(s); s != "" {
+			allowed[s] = true
+		}
+	}
+	return allowed
+}
+
+func firstConfigString(config *util.ViperProxy, keys ...string) string {
+	for _, key := range keys {
+		if v := config.GetString(key); v != "" {
+			return v
+		}
+	}
+	return ""
+}
+
+// matchesWildcardDomain requires a real subdomain of the wildcard: the name
+// must end with the domain, carry a non-empty label in front of it, and be a
+// valid DNS name, so a CA-signed name like ".example.com" cannot satisfy the
+// suffix check. A leading "*." label is allowed so wildcard certs keep
+// matching. One trailing root dot is ignored on both sides so absolute DNS
+// names still match.
+func matchesWildcardDomain(name, wildcardDomain string) bool {
+	name = strings.TrimSuffix(name, ".")
+	wildcardDomain = strings.TrimSuffix(wildcardDomain, ".")
+	if wildcardDomain == "" || len(name) <= len(wildcardDomain) || !strings.HasSuffix(name, wildcardDomain) {
+		return false
+	}
+	if wildcardDomain[0] != '.' && name[len(name)-len(wildcardDomain)-1] != '.' {
+		return false
+	}
+	return isDomainName(strings.TrimPrefix(name, "*."))
+}
+
+// isDomainName mirrors the Rust volume's check: non-empty labels of letters,
+// digits, '-', '_', each up to 63 chars and not '-'-edged, with at least one
+// non-digit so a bare IP cannot pass a domain wildcard.
+func isDomainName(name string) bool {
+	if len(name) == 0 || len(name) > 253 {
+		return false
+	}
+	nonNumeric := false
+	for _, label := range strings.Split(name, ".") {
+		if len(label) == 0 || len(label) > 63 || label[0] == '-' || label[len(label)-1] == '-' {
+			return false
+		}
+		for i := 0; i < len(label); i++ {
+			c := label[i]
+			if c != '-' && c != '_' && !(c >= '0' && c <= '9') && !(c >= 'a' && c <= 'z') && !(c >= 'A' && c <= 'Z') {
+				return false
+			}
+			nonNumeric = nonNumeric || c < '0' || c > '9'
+		}
+	}
+	return nonNumeric
+}
+
 func (a Authenticator) Authenticate(params *advancedtls.HandshakeVerificationInfo) (*advancedtls.PostHandshakeVerificationResults, error) {
-	if a.AllowedWildcardDomain != "" && strings.HasSuffix(params.Leaf.Subject.CommonName, a.AllowedWildcardDomain) {
+	if matchesWildcardDomain(params.Leaf.Subject.CommonName, a.AllowedWildcardDomain) {
 		return &advancedtls.PostHandshakeVerificationResults{}, nil
 	}
 	if _, ok := a.AllowedCommonNames[params.Leaf.Subject.CommonName]; ok {
@@ -284,6 +418,29 @@ func (a Authenticator) Authenticate(params *advancedtls.HandshakeVerificationInf
 	return nil, err
 }
 
+func (a Authenticator) AuthenticateServer(params *advancedtls.HandshakeVerificationInfo) (*advancedtls.PostHandshakeVerificationResults, error) {
+	leaf := params.Leaf
+	names := make([]string, 0, len(leaf.DNSNames)+len(leaf.URIs)+1)
+	names = append(names, leaf.Subject.CommonName)
+	names = append(names, leaf.DNSNames...)
+	for _, name := range names {
+		if a.AllowedCommonNames[name] || matchesWildcardDomain(name, a.AllowedWildcardDomain) {
+			return &advancedtls.PostHandshakeVerificationResults{}, nil
+		}
+	}
+	for _, u := range leaf.URIs {
+		// URI identities match exactly: a path that happens to end in the
+		// wildcard domain must not pass the DNS suffix check.
+		if a.AllowedCommonNames[u.String()] {
+			return &advancedtls.PostHandshakeVerificationResults{}, nil
+		}
+		names = append(names, u.String())
+	}
+	err := fmt.Errorf("AuthenticateServer: untrusted server identity %v", names)
+	glog.Error(err)
+	return nil, err
+}
+
 func FixTlsConfig(viper *util.ViperProxy, config *tls.Config) error {
 	var err error
 	config.MinVersion, err = TlsVersionByName(viper.GetString("tls.min_version"))
diff --git a/weed/security/tls_client_cert_test.go b/weed/security/tls_client_cert_test.go
index 7147f2e32d..82cd59b83e 100644
--- a/weed/security/tls_client_cert_test.go
+++ b/weed/security/tls_client_cert_test.go
@@ -10,6 +10,7 @@ import (
 	"encoding/pem"
 	"math/big"
 	"net"
+	"net/url"
 	"os"
 	"path/filepath"
 	"testing"
@@ -59,6 +60,10 @@ func newTestCA(t *testing.T, dir string) *testCA {
 
 // issue creates a leaf certificate restricted to the given extended key usages.
 func (ca *testCA) issue(t *testing.T, dir, name string, ekus []x509.ExtKeyUsage) (certFile, keyFile string) {
+	return ca.issueWithSANs(t, dir, name, ekus, []string{"localhost"}, nil)
+}
+
+func (ca *testCA) issueWithSANs(t *testing.T, dir, name string, ekus []x509.ExtKeyUsage, dnsNames []string, uris []string) (certFile, keyFile string) {
 	t.Helper()
 	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
 	if err != nil {
@@ -72,7 +77,14 @@ func (ca *testCA) issue(t *testing.T, dir, name string, ekus []x509.ExtKeyUsage)
 		KeyUsage:     x509.KeyUsageDigitalSignature,
 		ExtKeyUsage:  ekus,
 		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
-		DNSNames:     []string{"localhost"},
+		DNSNames:     dnsNames,
+	}
+	for _, u := range uris {
+		parsed, err := url.Parse(u)
+		if err != nil {
+			t.Fatal(err)
+		}
+		template.URIs = append(template.URIs, parsed)
 	}
 	der, err := x509.CreateCertificate(rand.Reader, template, ca.cert, &key.PublicKey, ca.key)
 	if err != nil {
@@ -181,3 +193,341 @@ func TestLoadClientTLSServerOnlyEkuRejected(t *testing.T) {
 		t.Fatal("expected handshake failure when presenting a serverAuth-only cert as client identity")
 	}
 }
+
+func TestLoadServerTLSUnconfiguredIsPlaintext(t *testing.T) {
+	v := util.NewViperProxy(viper.New())
+	creds, _, err := loadServerTLS(v, "grpc.master")
+	if err != nil || creds != nil {
+		t.Fatalf("expected no TLS and no error, got creds=%v err=%v", creds, err)
+	}
+}
+
+func TestLoadServerTLSPartialCertKeyFails(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	certFile, _ := ca.issue(t, dir, "server", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", certFile)
+
+	if _, _, err := loadServerTLS(v, "grpc.master"); err == nil {
+		t.Fatal("expected error when only cert is set")
+	}
+}
+
+func TestLoadServerTLSBadVersionNameFails(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	certFile, keyFile := ca.issue(t, dir, "server", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", certFile)
+	v.Set("grpc.master.key", keyFile)
+	v.Set("tls.min_version", "bogus")
+
+	if _, _, err := loadServerTLS(v, "grpc.master"); err == nil {
+		t.Fatal("expected error for unparseable tls.min_version")
+	}
+}
+
+func TestLoadClientTLSUnconfiguredIsPlaintext(t *testing.T) {
+	v := util.NewViperProxy(viper.New())
+	if _, err := loadClientTLS(v, "grpc.client", "grpc.client"); err != nil {
+		t.Fatalf("expected insecure dial option, got err=%v", err)
+	}
+}
+
+// A dedicated client pair missing one half is a misconfiguration: the dialer
+// must fail rather than silently dial with the fallback serving identity.
+func TestLoadClientTLSPartialClientPairFails(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	certFile, keyFile := ca.issue(t, dir, "dual", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth})
+	clientCert, _ := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.client.cert", certFile)
+	v.Set("grpc.client.key", keyFile)
+	v.Set("grpc.client.client_cert", clientCert)
+
+	if _, err := loadClientTLS(v, "grpc.client", "grpc.client"); err == nil {
+		t.Fatal("expected error for a partial client_cert/client_key pair with a valid fallback")
+	}
+}
+
+func TestLoadClientTLSMissingCaFails(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	certFile, keyFile := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.client.cert", certFile)
+	v.Set("grpc.client.key", keyFile)
+
+	if _, err := loadClientTLS(v, "grpc.client", "grpc.client"); err == nil {
+		t.Fatal("expected error when cert and key are set without grpc.ca")
+	}
+}
+
+func TestLoadClientTLSBadCaPathFails(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	certFile, keyFile := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", filepath.Join(dir, "missing-ca.pem"))
+	v.Set("grpc.client.cert", certFile)
+	v.Set("grpc.client.key", keyFile)
+
+	if _, err := loadClientTLS(v, "grpc.client", "grpc.client"); err == nil {
+		t.Fatal("expected error for unreadable grpc.ca")
+	}
+}
+
+func TestLoadClientTLSAllowsListedServerName(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issue(t, dir, "seaweed-filer", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", serverCert)
+	v.Set("grpc.master.key", serverKey)
+	v.Set("grpc.master.client_cert", clientCert)
+	v.Set("grpc.master.client_key", clientKey)
+	v.Set("grpc.master.allowed_server_commonNames", "seaweed-filer,seaweed-volume")
+
+	addr := startTestGrpcServer(t, v, "grpc.master")
+	if err := healthCheck(t, addr, LoadClientTLS(v, "grpc.master")); err != nil {
+		t.Fatalf("health check with listed server name failed: %v", err)
+	}
+}
+
+// Any CA-signed cert must not be able to stand in for another component:
+// a filer's identity dialing as "the master" has to fail the allowlist.
+func TestLoadClientTLSRejectsUnlistedServerName(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issue(t, dir, "seaweed-filer", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", serverCert)
+	v.Set("grpc.master.key", serverKey)
+	v.Set("grpc.master.client_cert", clientCert)
+	v.Set("grpc.master.client_key", clientKey)
+	v.Set("grpc.master.allowed_server_commonNames", "seaweed-master")
+
+	addr := startTestGrpcServer(t, v, "grpc.master")
+	if err := healthCheck(t, addr, LoadClientTLS(v, "grpc.master")); err == nil {
+		t.Fatal("expected handshake failure for a server identity outside the allowlist")
+	}
+}
+
+func TestLoadClientTLSMatchesServerURISAN(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issueWithSANs(t, dir, "server",
+		[]x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, nil, []string{"spiffe://cluster.local/ns/seaweedfs/master"})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", serverCert)
+	v.Set("grpc.master.key", serverKey)
+	v.Set("grpc.master.client_cert", clientCert)
+	v.Set("grpc.master.client_key", clientKey)
+	v.Set("grpc.master.allowed_server_commonNames", "spiffe://cluster.local/ns/seaweedfs/master")
+
+	addr := startTestGrpcServer(t, v, "grpc.master")
+	if err := healthCheck(t, addr, LoadClientTLS(v, "grpc.master")); err != nil {
+		t.Fatalf("health check with URI SAN allowlist failed: %v", err)
+	}
+}
+
+func TestLoadClientTLSMatchesServerWildcardDomain(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issueWithSANs(t, dir, "server",
+		[]x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, []string{"master.example.com"}, nil)
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", serverCert)
+	v.Set("grpc.master.key", serverKey)
+	v.Set("grpc.master.client_cert", clientCert)
+	v.Set("grpc.master.client_key", clientKey)
+	v.Set("grpc.allowed_server_wildcard_domain", ".example.com")
+
+	addr := startTestGrpcServer(t, v, "grpc.master")
+	if err := healthCheck(t, addr, LoadClientTLS(v, "grpc.master")); err != nil {
+		t.Fatalf("health check with wildcard domain failed: %v", err)
+	}
+}
+
+func TestMatchesWildcardDomain(t *testing.T) {
+	cases := []struct {
+		name   string
+		domain string
+		want   bool
+	}{
+		{"node.prod.example.com", ".example.com", true},
+		{"node.example.com.", ".example.com", true},
+		{"node.example.com.", ".example.com.", true},
+		{"node.example.com..", ".example.com", false},
+		{"node..example.com.", ".example.com", false},
+		{"example.com", ".example.com", false},
+		{".example.com", ".example.com", false},
+		{"node..example.com", ".example.com", false},
+		{"node-.example.com", ".example.com", false},
+		{"-node.example.com", ".example.com", false},
+		{"node.example.com", "", false},
+		{"node.example.com", ".other.net", false},
+		{"badexample.com", "example.com", false},
+		{"good.example.com", "example.com", true},
+		{"*.example.com", ".example.com", true},
+		{"1.2.3.4", ".3.4", false},
+	}
+	for _, c := range cases {
+		if got := matchesWildcardDomain(c.name, c.domain); got != c.want {
+			t.Errorf("matchesWildcardDomain(%q, %q) = %v, want %v", c.name, c.domain, got, c.want)
+		}
+	}
+}
+
+// A CA-signed name that is not a valid hostname must not satisfy the wildcard
+// suffix check: ".example.com" equals the wildcard and has no hostname label.
+func TestLoadClientTLSRejectsInvalidWildcardName(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issueWithSANs(t, dir, ".example.com",
+		[]x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, []string{".example.com"}, nil)
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", serverCert)
+	v.Set("grpc.master.key", serverKey)
+	v.Set("grpc.master.client_cert", clientCert)
+	v.Set("grpc.master.client_key", clientKey)
+	v.Set("grpc.allowed_server_wildcard_domain", ".example.com")
+
+	addr := startTestGrpcServer(t, v, "grpc.master")
+	if err := healthCheck(t, addr, LoadClientTLS(v, "grpc.master")); err == nil {
+		t.Fatal("expected handshake failure: a non-hostname identity must not match the wildcard domain")
+	}
+}
+
+// A URI SAN whose path happens to end in the wildcard domain must not pass:
+// URI identities are exact-match only.
+func TestLoadClientTLSURIPathDoesNotMatchWildcardDomain(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issueWithSANs(t, dir, "server",
+		[]x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, nil, []string{"spiffe://other.invalid/workload.example.com"})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", serverCert)
+	v.Set("grpc.master.key", serverKey)
+	v.Set("grpc.master.client_cert", clientCert)
+	v.Set("grpc.master.client_key", clientKey)
+	v.Set("grpc.allowed_server_wildcard_domain", ".example.com")
+
+	addr := startTestGrpcServer(t, v, "grpc.master")
+	if err := healthCheck(t, addr, LoadClientTLS(v, "grpc.master")); err == nil {
+		t.Fatal("expected handshake failure: URI SAN path must not match the wildcard domain")
+	}
+}
+
+func TestLoadClientTLSGlobalServerNamesFallback(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issue(t, dir, "seaweed-filer", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.master.cert", serverCert)
+	v.Set("grpc.master.key", serverKey)
+	v.Set("grpc.master.client_cert", clientCert)
+	v.Set("grpc.master.client_key", clientKey)
+	v.Set("grpc.allowed_server_commonNames", "seaweed-filer")
+
+	addr := startTestGrpcServer(t, v, "grpc.master")
+	if err := healthCheck(t, addr, LoadClientTLS(v, "grpc.master")); err != nil {
+		t.Fatalf("health check with global allowlist failed: %v", err)
+	}
+}
+
+// A per-target allowlist lives under the dialing component: the filer reads
+// grpc.filer.volume_allowed_server_commonNames for volume calls, never
+// grpc.volume.allowed_server_commonNames, which is the volume's own outbound
+// allowlist (e.g. naming only the master).
+func TestLoadClientTLSForTargetUsesClientScopedAllowlist(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issue(t, dir, "seaweed-volume", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.volume.cert", serverCert)
+	v.Set("grpc.volume.key", serverKey)
+	v.Set("grpc.volume.allowed_server_commonNames", "seaweed-master")
+	v.Set("grpc.filer.client_cert", clientCert)
+	v.Set("grpc.filer.client_key", clientKey)
+	v.Set("grpc.filer.volume_allowed_server_commonNames", "seaweed-volume")
+
+	addr := startTestGrpcServer(t, v, "grpc.volume")
+	if err := healthCheck(t, addr, LoadClientTLSForTarget(v, "grpc.filer", "grpc.volume")); err != nil {
+		t.Fatalf("health check with client-scoped target allowlist failed: %v", err)
+	}
+}
+
+func TestLoadClientTLSForTargetIgnoresTargetOutboundAllowlist(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issue(t, dir, "seaweed-volume", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.volume.cert", serverCert)
+	v.Set("grpc.volume.key", serverKey)
+	v.Set("grpc.volume.allowed_server_commonNames", "seaweed-master")
+	v.Set("grpc.filer.client_cert", clientCert)
+	v.Set("grpc.filer.client_key", clientKey)
+
+	addr := startTestGrpcServer(t, v, "grpc.volume")
+	if err := healthCheck(t, addr, LoadClientTLSForTarget(v, "grpc.filer", "grpc.volume")); err != nil {
+		t.Fatalf("health check must not apply the target's outbound allowlist: %v", err)
+	}
+}
+
+func TestLoadClientTLSForTargetFallsBackToClientAllowlist(t *testing.T) {
+	dir := t.TempDir()
+	ca := newTestCA(t, dir)
+	serverCert, serverKey := ca.issue(t, dir, "seaweed-volume", []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth})
+	clientCert, clientKey := ca.issue(t, dir, "client", []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth})
+
+	v := util.NewViperProxy(viper.New())
+	v.Set("grpc.ca", ca.file)
+	v.Set("grpc.volume.cert", serverCert)
+	v.Set("grpc.volume.key", serverKey)
+	v.Set("grpc.filer.client_cert", clientCert)
+	v.Set("grpc.filer.client_key", clientKey)
+	v.Set("grpc.filer.allowed_server_commonNames", "seaweed-volume")
+
+	addr := startTestGrpcServer(t, v, "grpc.volume")
+	if err := healthCheck(t, addr, LoadClientTLSForTarget(v, "grpc.filer", "grpc.volume")); err != nil {
+		t.Fatalf("health check with client component allowlist failed: %v", err)
+	}
+}
diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go
index 498bdd624c..eb3b455b79 100644
--- a/weed/server/filer_grpc_server.go
+++ b/weed/server/filer_grpc_server.go
@@ -1237,7 +1237,7 @@ func (fs *FilerServer) AssignVolume(ctx context.Context, req *filer_pb.AssignVol
 		altRequest.ExpectedDataSize = req.ExpectedDataSize
 	}
 
-	assignResult, err := operation.Assign(ctx, fs.filer.GetMaster, fs.grpcDialOption, assignRequest, altRequest)
+	assignResult, err := operation.Assign(ctx, fs.filer.GetMaster, fs.masterGrpcDialOption(), assignRequest, altRequest)
 	if err != nil {
 		glog.V(3).InfofCtx(ctx, "AssignVolume: %v", err)
 		return &filer_pb.AssignVolumeResponse{Error: fmt.Sprintf("assign volume: %v", err)}, nil
diff --git a/weed/server/filer_grpc_server_admin.go b/weed/server/filer_grpc_server_admin.go
index c013cff2e8..14a73265cf 100644
--- a/weed/server/filer_grpc_server_admin.go
+++ b/weed/server/filer_grpc_server_admin.go
@@ -121,7 +121,7 @@ func (fs *FilerServer) Ping(ctx context.Context, req *filer_pb.PingRequest) (res
 		})
 	}
 	if req.TargetType == cluster.VolumeServerType {
-		pingErr = pb.WithVolumeServerClient(false, pb.ServerAddress(req.Target), fs.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
+		pingErr = pb.WithVolumeServerClient(false, pb.ServerAddress(req.Target), fs.volumeGrpcDialOption(), func(client volume_server_pb.VolumeServerClient) error {
 			pingResp, err := client.Ping(ctx, &volume_server_pb.PingRequest{})
 			if pingResp != nil {
 				resp.RemoteTimeNs = pingResp.StartTimeNs
@@ -130,7 +130,7 @@ func (fs *FilerServer) Ping(ctx context.Context, req *filer_pb.PingRequest) (res
 		})
 	}
 	if req.TargetType == cluster.MasterType {
-		pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), fs.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
+		pingErr = pb.WithMasterClient(context.Background(), false, pb.ServerAddress(req.Target), fs.masterGrpcDialOption(), false, func(client master_pb.SeaweedClient) error {
 			pingResp, err := client.Ping(ctx, &master_pb.PingRequest{})
 			if pingResp != nil {
 				resp.RemoteTimeNs = pingResp.StartTimeNs
diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go
index 78f5e1f610..22597a71cf 100644
--- a/weed/server/filer_grpc_server_remote.go
+++ b/weed/server/filer_grpc_server_remote.go
@@ -205,7 +205,7 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re
 			}
 
 			// assign one volume server
-			assignResult, err := operation.Assign(ctx, fs.filer.GetMaster, fs.grpcDialOption, assignRequest, altRequest)
+			assignResult, err := operation.Assign(ctx, fs.filer.GetMaster, fs.masterGrpcDialOption(), assignRequest, altRequest)
 			if err != nil {
 				chunksMu.Lock()
 				if fetchAndWriteErr == nil {
@@ -244,7 +244,7 @@ func (fs *FilerServer) doCacheRemoteObjectToLocalCluster(ctx context.Context, re
 			// tell filer to tell volume server to download into needles
 			assignedServerAddress := pb.NewServerAddressWithGrpcPort(assignResult.Url, assignResult.GrpcPort)
 			var etag string
-			err = operation.WithVolumeServerClient(false, assignedServerAddress, fs.grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {
+			err = operation.WithVolumeServerClient(false, assignedServerAddress, fs.volumeGrpcDialOption(), func(volumeServerClient volume_server_pb.VolumeServerClient) error {
 				resp, fetchErr := volumeServerClient.FetchAndWriteNeedle(context.Background(), &volume_server_pb.FetchAndWriteNeedleRequest{
 					VolumeId:            uint32(fileId.VolumeId),
 					NeedleId:            uint64(fileId.Key),
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 3f8b0a9e95..bfbbcf1bd2 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -123,11 +123,14 @@ type FilerServer struct {
 	inFlightDataLimitCond *sync.Cond
 
 	filer_pb.UnimplementedSeaweedFilerServer
-	option         *FilerOption
-	filer          *filer.Filer
-	filerGuard     *security.Guard
-	volumeGuard    *security.Guard
-	grpcDialOption grpc.DialOption
+	option               *FilerOption
+	filer                *filer.Filer
+	filerGuard           *security.Guard
+	volumeGuard          *security.Guard
+	grpcDialOption       grpc.DialOption
+	grpcDialOptionMaster grpc.DialOption
+	grpcDialOptionVolume grpc.DialOption
+	grpcDialOptionS3     grpc.DialOption
 
 	// metrics read from the master
 	metricsAddress     string
@@ -258,6 +261,9 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	fs = &FilerServer{
 		option:                option,
 		grpcDialOption:        security.LoadClientTLS(util.GetViper(), "grpc.filer"),
+		grpcDialOptionMaster:  security.LoadClientTLSForTarget(util.GetViper(), "grpc.filer", "grpc.master"),
+		grpcDialOptionVolume:  security.LoadClientTLSForTarget(util.GetViper(), "grpc.filer", "grpc.volume"),
+		grpcDialOptionS3:      security.LoadClientTLSForTarget(util.GetViper(), "grpc.filer", "grpc.s3"),
 		knownListeners:        make(map[int32]int32),
 		subscribers:           make(map[int32]*metadataSubscriber),
 		inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
@@ -294,7 +300,11 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 	v.SetDefault("filer.options.max_file_name_length", 255)
 	maxFilenameLength := v.GetUint32("filer.options.max_file_name_length")
 	glog.V(0).Infof("max_file_name_length %d", maxFilenameLength)
-	fs.filer = filer.NewFiler(*option.Masters, fs.grpcDialOption, option.Host, option.FilerGroup, option.Collection, option.DefaultReplication, option.DataCenter, maxFilenameLength, nil)
+	fs.filer = filer.NewFiler(*option.Masters, &filer.DialOptions{
+		Filer:  fs.grpcDialOption,
+		Master: fs.masterGrpcDialOption(),
+		Volume: fs.volumeGrpcDialOption(),
+	}, option.Host, option.FilerGroup, option.Collection, option.DefaultReplication, option.DataCenter, maxFilenameLength, nil)
 	// Peer metadata replay takes the same per-path locks, so a replayed change
 	// cannot split a local mutation's read-modify-write.
 	fs.entryLockTable = fs.filer.EntryLockTable
@@ -423,19 +433,42 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		fs.CredentialManager.SetFilerAddressFunc(func() pb.ServerAddress {
 			return fs.option.Host
 		}, fs.grpcDialOption)
-		fs.CredentialManager.SetMasterClient(fs.filer.MasterClient, fs.grpcDialOption)
+		fs.CredentialManager.SetMasterClient(fs.filer.MasterClient, fs.s3GrpcDialOption())
 	}
 
 	return fs, nil
 }
 
+// Target-specific dial options fall back to the shared filer option, which is
+// what FilerServer literals in tests set up.
+func (fs *FilerServer) masterGrpcDialOption() grpc.DialOption {
+	if fs.grpcDialOptionMaster != nil {
+		return fs.grpcDialOptionMaster
+	}
+	return fs.grpcDialOption
+}
+
+func (fs *FilerServer) volumeGrpcDialOption() grpc.DialOption {
+	if fs.grpcDialOptionVolume != nil {
+		return fs.grpcDialOptionVolume
+	}
+	return fs.grpcDialOption
+}
+
+func (fs *FilerServer) s3GrpcDialOption() grpc.DialOption {
+	if fs.grpcDialOptionS3 != nil {
+		return fs.grpcDialOptionS3
+	}
+	return fs.grpcDialOption
+}
+
 func (fs *FilerServer) checkWithMaster() {
 
 	isConnected := false
 	for !isConnected {
 		fs.option.Masters.RefreshBySrvIfAvailable()
 		for _, master := range fs.option.Masters.GetInstances() {
-			readErr := operation.WithMasterServerClient(context.Background(), false, master, fs.grpcDialOption, func(masterClient master_pb.SeaweedClient) error {
+			readErr := operation.WithMasterServerClient(context.Background(), false, master, fs.masterGrpcDialOption(), func(masterClient master_pb.SeaweedClient) error {
 				resp, err := masterClient.GetMasterConfiguration(context.Background(), &master_pb.GetMasterConfigurationRequest{})
 				if err != nil {
 					return fmt.Errorf("get master %s configuration: %v", master, err)
@@ -655,7 +688,7 @@ func (fs *FilerServer) waitForClusterRemoval() {
 			// The per-query timeout shares the overall deadline, so a stalled
 			// master cannot stretch the wait past the budget.
 			queryCtx, queryCancel := context.WithTimeout(ctx, 2*time.Second)
-			nodes, err := cluster.ListExistingPeerUpdates(queryCtx, master, fs.grpcDialOption, fs.filer.MasterClient.FilerGroup, cluster.FilerType)
+			nodes, err := cluster.ListExistingPeerUpdates(queryCtx, master, fs.masterGrpcDialOption(), fs.filer.MasterClient.FilerGroup, cluster.FilerType)
 			queryCancel()
 			if err != nil {
 				// A failed query is not confirmed removal; retry until the
diff --git a/weed/server/filer_server_handlers_copy.go b/weed/server/filer_server_handlers_copy.go
index 9db334b1de..ed869dd617 100644
--- a/weed/server/filer_server_handlers_copy.go
+++ b/weed/server/filer_server_handlers_copy.go
@@ -732,7 +732,7 @@ func (fs *FilerServer) batchLookupVolumeLocations(ctx context.Context, chunks []
 	}
 
 	// Perform single batched lookup
-	lookupResult, err := operation.LookupVolumeIds(fs.filer.GetMaster, fs.grpcDialOption, volumeIdStrs)
+	lookupResult, err := operation.LookupVolumeIds(fs.filer.GetMaster, fs.masterGrpcDialOption(), volumeIdStrs)
 	if err != nil {
 		return nil, fmt.Errorf("failed to lookup volumes: %w", err)
 	}
diff --git a/weed/server/filer_server_handlers_write.go b/weed/server/filer_server_handlers_write.go
index f4208168cf..28f1b61f3b 100644
--- a/weed/server/filer_server_handlers_write.go
+++ b/weed/server/filer_server_handlers_write.go
@@ -55,7 +55,7 @@ func (fs *FilerServer) assignNewFileInfo(ctx context.Context, so *operation.Stor
 	// Use a context that ignores cancellation from the request context
 	assignCtx := context.WithoutCancel(ctx)
 
-	assignResult, ae := operation.Assign(assignCtx, fs.filer.GetMaster, fs.grpcDialOption, ar, altRequest)
+	assignResult, ae := operation.Assign(assignCtx, fs.filer.GetMaster, fs.masterGrpcDialOption(), ar, altRequest)
 	if ae != nil {
 		glog.ErrorfCtx(ctx, "failing to assign a file id: %v", ae)
 		err = ae