This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-kubernetes.git
The following commit(s) were added to refs/heads/master by this push:
new 299a8c77 Fixed grpc inbound potential issues (#987)
299a8c77 is described below
commit 299a8c7738c854bea618bc3ee1f78a2cfc18829f
Author: mfordjody <[email protected]>
AuthorDate: Thu Aug 6 17:28:26 2026 +0800
Fixed grpc inbound potential issues (#987)
---
dubbod/discovery/cmd/app/grpc_inbound.go | 359 +++++++++++++++++++++++---
dubbod/discovery/cmd/app/grpc_inbound_test.go | 274 +++++++++++++++++++-
2 files changed, 584 insertions(+), 49 deletions(-)
diff --git a/dubbod/discovery/cmd/app/grpc_inbound.go
b/dubbod/discovery/cmd/app/grpc_inbound.go
index f31736b9..17c86954 100644
--- a/dubbod/discovery/cmd/app/grpc_inbound.go
+++ b/dubbod/discovery/cmd/app/grpc_inbound.go
@@ -24,6 +24,7 @@ import (
"fmt"
"io"
"net"
+ neturl "net/url"
"os"
"strconv"
"strings"
@@ -37,13 +38,16 @@ import (
)
type grpcInboundOptions struct {
- listen string
- upstream string
- bootstrapPath string
- runtimeConfig string
- mtlsMode string
- acceptTimeout time.Duration
- connectTimeout time.Duration
+ listen string
+ upstream string
+ bootstrapPath string
+ runtimeConfig string
+ mtlsMode string
+ trustDomain string
+ allowedPrincipals string
+ acceptTimeout time.Duration
+ connectTimeout time.Duration
+ reloadInterval time.Duration
}
type grpcInboundMTLSMode string
@@ -54,15 +58,30 @@ const (
grpcInboundMTLSModeStrict grpcInboundMTLSMode = "STRICT"
)
+// grpcInboundReloadInterval bounds how stale the workload certificate and the
+// runtime config may be. Certificates are rotated well ahead of expiry by the
+// control plane, so polling is enough and avoids the inotify blind spot around
+// kubelet's atomic symlink swap of the mounted secret.
+const grpcInboundReloadInterval = 30 * time.Second
+
+// grpcInboundAcceptTimeout caps how long a peer may take to get through the
+// first read and the TLS handshake. Every mesh pod can reach this port, so
+// without a deadline a peer that connects and never writes pins a goroutine
+// and a file descriptor indefinitely.
+const grpcInboundAcceptTimeout = 10 * time.Second
+
func newGRPCInboundCommand() *cobra.Command {
opts := &grpcInboundOptions{
- listen:
firstNonEmpty(os.Getenv("DUBBO_GRPC_INBOUND_LISTEN"), fmt.Sprintf(":%d",
inject.ProxylessGRPCInboundPort)),
- upstream:
firstNonEmpty(os.Getenv("DUBBO_GRPC_INBOUND_UPSTREAM"), "127.0.0.1:80"),
- bootstrapPath: os.Getenv("GRPC_XDS_BOOTSTRAP"),
- runtimeConfig:
firstNonEmpty(os.Getenv(inject.ProxylessGRPCConfigEnvName),
inject.ProxylessGRPCConfigPath),
- mtlsMode: os.Getenv("DUBBO_GRPC_INBOUND_MTLS_MODE"),
- acceptTimeout:
durationSecondsFromEnv("DUBBO_GRPC_INBOUND_ACCEPT_TIMEOUT", 0),
- connectTimeout:
durationSecondsFromEnv("DUBBO_GRPC_INBOUND_CONNECT_TIMEOUT", 5*time.Second),
+ listen:
firstNonEmpty(os.Getenv("DUBBO_GRPC_INBOUND_LISTEN"), fmt.Sprintf(":%d",
inject.ProxylessGRPCInboundPort)),
+ upstream:
firstNonEmpty(os.Getenv("DUBBO_GRPC_INBOUND_UPSTREAM"), "127.0.0.1:80"),
+ bootstrapPath: os.Getenv("GRPC_XDS_BOOTSTRAP"),
+ runtimeConfig:
firstNonEmpty(os.Getenv(inject.ProxylessGRPCConfigEnvName),
inject.ProxylessGRPCConfigPath),
+ mtlsMode: os.Getenv("DUBBO_GRPC_INBOUND_MTLS_MODE"),
+ trustDomain:
firstNonEmpty(os.Getenv("DUBBO_GRPC_INBOUND_TRUST_DOMAIN"),
os.Getenv("TRUST_DOMAIN")),
+ allowedPrincipals:
os.Getenv("DUBBO_GRPC_INBOUND_ALLOWED_PRINCIPALS"),
+ acceptTimeout:
durationSecondsFromEnv("DUBBO_GRPC_INBOUND_ACCEPT_TIMEOUT",
grpcInboundAcceptTimeout),
+ connectTimeout:
durationSecondsFromEnv("DUBBO_GRPC_INBOUND_CONNECT_TIMEOUT", 5*time.Second),
+ reloadInterval:
durationSecondsFromEnv("DUBBO_GRPC_INBOUND_RELOAD_INTERVAL",
grpcInboundReloadInterval),
}
c := &cobra.Command{
Use: "grpc-inbound",
@@ -81,8 +100,12 @@ func newGRPCInboundCommand() *cobra.Command {
c.Flags().StringVar(&opts.bootstrapPath, "bootstrap",
opts.bootstrapPath, "gRPC xDS bootstrap file")
c.Flags().StringVar(&opts.runtimeConfig, "runtime-config",
opts.runtimeConfig, "proxyless runtime config file")
c.Flags().StringVar(&opts.mtlsMode, "mtls-mode", opts.mtlsMode,
"override inbound mTLS mode: DISABLE, PERMISSIVE, or STRICT")
- c.Flags().DurationVar(&opts.acceptTimeout, "accept-timeout",
opts.acceptTimeout, "optional TLS handshake timeout")
+ c.Flags().StringVar(&opts.trustDomain, "trust-domain",
opts.trustDomain, "trust domain peers must belong to; defaults to the trust
domain of the workload certificate")
+ c.Flags().StringVar(&opts.allowedPrincipals, "allowed-principals",
opts.allowedPrincipals,
+ "comma-separated peer identities allowed to connect, as
spiffe:// URIs or ns/<namespace>/sa/<serviceaccount>; empty allows any peer in
the trust domain")
+ c.Flags().DurationVar(&opts.acceptTimeout, "accept-timeout",
opts.acceptTimeout, "deadline for the first read and the TLS handshake; 0
disables it")
c.Flags().DurationVar(&opts.connectTimeout, "connect-timeout",
opts.connectTimeout, "timeout for connecting to the local upstream")
+ c.Flags().DurationVar(&opts.reloadInterval, "reload-interval",
opts.reloadInterval, "how often to reload the workload certificate and runtime
config")
return c
}
@@ -100,7 +123,18 @@ func (o *grpcInboundOptions) run(ctx context.Context)
error {
if err != nil {
return err
}
- tlsConfig, err := grpcInboundTLSConfigFromBootstrap(bootstrap)
+ certs, err := newGRPCInboundCertStore(bootstrap)
+ if err != nil {
+ return err
+ }
+ modes := newGRPCInboundModeStore(o.runtimeConfig,
upstreamPort(o.upstream))
+ if err := modes.reload(); err != nil {
+ // A missing or unreadable runtime config is not fatal: the
store falls
+ // back to STRICT until a successful load, and an explicit
--mtls-mode
+ // still overrides it.
+ log.Warnf("grpc-inbound: initial runtime config load failed:
%v", err)
+ }
+ peers, err := o.peerPolicy(certs)
if err != nil {
return err
}
@@ -109,10 +143,61 @@ func (o *grpcInboundOptions) run(ctx context.Context)
error {
return fmt.Errorf("listen grpc-inbound %s: %w", o.listen, err)
}
defer lis.Close()
- return serveGRPCInbound(ctx, lis, tlsConfig, o.upstream,
o.effectiveMTLSMode, o.acceptTimeout, o.connectTimeout)
+ go grpcInboundReloadLoop(ctx, o.reloadInterval, certs, modes)
+ return serveGRPCInbound(ctx, lis, certs.tlsConfig(peers), o.upstream,
o.effectiveMTLSMode(modes), o.acceptTimeout, o.connectTimeout)
+}
+
+func (o *grpcInboundOptions) peerPolicy(certs *grpcInboundCertStore)
(*grpcInboundPeerPolicy, error) {
+ trustDomain := firstNonEmpty(strings.TrimSpace(o.trustDomain),
certs.trustDomain())
+ allowed, err := parseGRPCInboundPrincipals(o.allowedPrincipals,
trustDomain)
+ if err != nil {
+ return nil, err
+ }
+ if trustDomain == "" {
+ log.Warnf("grpc-inbound: no trust domain configured and the
workload certificate carries no SPIFFE identity; peer identity is not checked")
+ }
+ return &grpcInboundPeerPolicy{trustDomain: trustDomain, allowed:
allowed}, nil
}
-func grpcInboundTLSConfigFromBootstrap(bootstrap *xdsresolver.BootstrapConfig)
(*tls.Config, error) {
+// grpcInboundReloadLoop keeps the workload certificate and the runtime config
+// in sync with the mounted secret. Without it the process would serve the
+// key pair it loaded at startup until the pod is restarted, so inbound mTLS
+// would break as soon as the control plane rotates the certificate.
+func grpcInboundReloadLoop(ctx context.Context, interval time.Duration, certs
*grpcInboundCertStore, modes *grpcInboundModeStore) {
+ if interval <= 0 {
+ interval = grpcInboundReloadInterval
+ }
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ if err := certs.reload(); err != nil {
+ log.Warnf("grpc-inbound: certificate reload
failed, keeping previous material: %v", err)
+ }
+ if err := modes.reload(); err != nil {
+ log.Warnf("grpc-inbound: runtime config reload
failed, keeping previous mode: %v", err)
+ }
+ }
+ }
+}
+
+// grpcInboundCertStore holds the currently loaded workload key pair and trust
+// bundle. Handshakes read through it, so a reload takes effect on the next
+// connection without dropping established ones.
+type grpcInboundCertStore struct {
+ certFile string
+ keyFile string
+ caFile string
+
+ mu sync.RWMutex
+ cert *tls.Certificate
+ clientCAs *x509.CertPool
+}
+
+func newGRPCInboundCertStore(bootstrap *xdsresolver.BootstrapConfig)
(*grpcInboundCertStore, error) {
if bootstrap == nil {
return nil, fmt.Errorf("bootstrap config is nil")
}
@@ -126,24 +211,168 @@ func grpcInboundTLSConfigFromBootstrap(bootstrap
*xdsresolver.BootstrapConfig) (
if cfg.CACertificateFile == "" {
return nil, fmt.Errorf("grpc-inbound mTLS requires
ca_certificate_file")
}
- cert, err := tls.LoadX509KeyPair(cfg.CertificateFile,
cfg.PrivateKeyFile)
+ store := &grpcInboundCertStore{
+ certFile: cfg.CertificateFile,
+ keyFile: cfg.PrivateKeyFile,
+ caFile: cfg.CACertificateFile,
+ }
+ if err := store.reload(); err != nil {
+ return nil, err
+ }
+ return store, nil
+}
+
+func (s *grpcInboundCertStore) reload() error {
+ cert, err := tls.LoadX509KeyPair(s.certFile, s.keyFile)
if err != nil {
- return nil, fmt.Errorf("load grpc-inbound certificate/key: %w",
err)
+ return fmt.Errorf("load grpc-inbound certificate/key: %w", err)
}
- rootPEM, err := os.ReadFile(cfg.CACertificateFile)
+ // Leaf is needed to read the workload's own SPIFFE identity; older Go
+ // releases leave it nil after LoadX509KeyPair.
+ if cert.Leaf == nil && len(cert.Certificate) > 0 {
+ leaf, err := x509.ParseCertificate(cert.Certificate[0])
+ if err != nil {
+ return fmt.Errorf("parse grpc-inbound certificate %s:
%w", s.certFile, err)
+ }
+ cert.Leaf = leaf
+ }
+ rootPEM, err := os.ReadFile(s.caFile)
if err != nil {
- return nil, fmt.Errorf("read grpc-inbound CA certificate %s:
%w", cfg.CACertificateFile, err)
+ return fmt.Errorf("read grpc-inbound CA certificate %s: %w",
s.caFile, err)
}
clientCAs := x509.NewCertPool()
if !clientCAs.AppendCertsFromPEM(rootPEM) {
- return nil, fmt.Errorf("parse grpc-inbound CA certificate %s:
no certificates found", cfg.CACertificateFile)
+ return fmt.Errorf("parse grpc-inbound CA certificate %s: no
certificates found", s.caFile)
}
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.cert = &cert
+ s.clientCAs = clientCAs
+ return nil
+}
+
+func (s *grpcInboundCertStore) current() (*tls.Certificate, *x509.CertPool) {
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+ return s.cert, s.clientCAs
+}
+
+// tlsConfig returns a config whose material is resolved per handshake.
+// GetConfigForClient is used rather than GetCertificate because the trust
+// bundle rotates alongside the key pair and ClientCAs cannot be swapped from
+// the certificate callback.
+func (s *grpcInboundCertStore) tlsConfig(peers *grpcInboundPeerPolicy)
*tls.Config {
return &tls.Config{
- MinVersion: tls.VersionTLS12,
- Certificates: []tls.Certificate{cert},
- ClientCAs: clientCAs,
- ClientAuth: tls.RequireAndVerifyClientCert,
- }, nil
+ MinVersion: tls.VersionTLS12,
+ ClientAuth: tls.RequireAndVerifyClientCert,
+ GetConfigForClient: func(*tls.ClientHelloInfo) (*tls.Config,
error) {
+ cert, clientCAs := s.current()
+ if cert == nil || clientCAs == nil {
+ return nil, fmt.Errorf("grpc-inbound
certificate material is not loaded")
+ }
+ return &tls.Config{
+ MinVersion: tls.VersionTLS12,
+ Certificates: []tls.Certificate{*cert},
+ ClientCAs: clientCAs,
+ ClientAuth:
tls.RequireAndVerifyClientCert,
+ VerifyPeerCertificate:
peers.verifyPeerCertificate,
+ }, nil
+ },
+ }
+}
+
+// trustDomain reports the trust domain of the workload's own SPIFFE identity,
+// used as the default peer trust domain when none is configured explicitly.
+func (s *grpcInboundCertStore) trustDomain() string {
+ cert, _ := s.current()
+ if cert == nil || cert.Leaf == nil {
+ return ""
+ }
+ for _, id := range spiffeIdentities(cert.Leaf) {
+ return id.Host
+ }
+ return ""
+}
+
+// grpcInboundPeerPolicy authorizes an authenticated peer. Chain verification
+// alone only proves the peer holds a certificate signed by the mesh CA, which
+// makes every workload in the mesh a valid caller for every other workload.
+// This narrows that to a trust domain and, when configured, to an explicit set
+// of SPIFFE identities.
+type grpcInboundPeerPolicy struct {
+ trustDomain string
+ allowed map[string]struct{}
+}
+
+// verifyPeerCertificate runs after chain verification, so verifiedChains is
+// non-empty and its leaf is already trusted. It only decides whether that
+// proven identity may talk to this workload.
+func (p *grpcInboundPeerPolicy) verifyPeerCertificate(_ [][]byte,
verifiedChains [][]*x509.Certificate) error {
+ if p == nil || (p.trustDomain == "" && len(p.allowed) == 0) {
+ return nil
+ }
+ if len(verifiedChains) == 0 || len(verifiedChains[0]) == 0 {
+ return fmt.Errorf("grpc-inbound: peer presented no verified
certificate chain")
+ }
+ identities := spiffeIdentities(verifiedChains[0][0])
+ if len(identities) == 0 {
+ return fmt.Errorf("grpc-inbound: peer certificate carries no
SPIFFE identity")
+ }
+ for _, id := range identities {
+ if p.trustDomain != "" && id.Host != p.trustDomain {
+ continue
+ }
+ if len(p.allowed) == 0 {
+ return nil
+ }
+ if _, ok := p.allowed[id.String()]; ok {
+ return nil
+ }
+ }
+ return fmt.Errorf("grpc-inbound: peer identity %s is not authorized",
identities[0])
+}
+
+// spiffeIdentities returns the SPIFFE URI SANs of a certificate. A workload
+// certificate normally carries exactly one.
+func spiffeIdentities(cert *x509.Certificate) []*neturl.URL {
+ if cert == nil {
+ return nil
+ }
+ out := make([]*neturl.URL, 0, len(cert.URIs))
+ for _, uri := range cert.URIs {
+ if uri != nil && uri.Scheme == "spiffe" {
+ out = append(out, uri)
+ }
+ }
+ return out
+}
+
+// parseGRPCInboundPrincipals accepts full spiffe:// URIs or the
+// ns/<namespace>/sa/<serviceaccount> shorthand, which is expanded against the
+// trust domain.
+func parseGRPCInboundPrincipals(list, trustDomain string)
(map[string]struct{}, error) {
+ out := map[string]struct{}{}
+ for _, raw := range strings.Split(list, ",") {
+ entry := strings.TrimSpace(raw)
+ if entry == "" {
+ continue
+ }
+ if !strings.HasPrefix(entry, "spiffe://") {
+ if trustDomain == "" {
+ return nil, fmt.Errorf("principal %q needs a
trust domain: set --trust-domain or use a full spiffe:// URI", entry)
+ }
+ entry = "spiffe://" + trustDomain + "/" +
strings.TrimPrefix(entry, "/")
+ }
+ parsed, err := neturl.Parse(entry)
+ if err != nil {
+ return nil, fmt.Errorf("parse principal %q: %w", raw,
err)
+ }
+ out[parsed.String()] = struct{}{}
+ }
+ if len(out) == 0 {
+ return nil, nil
+ }
+ return out, nil
}
func serveGRPCInbound(ctx context.Context, lis net.Listener, tlsConfig
*tls.Config, upstream string, mode func() grpcInboundMTLSMode, acceptTimeout,
connectTimeout time.Duration) error {
@@ -216,14 +445,54 @@ func isTLSClientHello(first byte) bool {
return first == 0x16
}
-func (o *grpcInboundOptions) effectiveMTLSMode() grpcInboundMTLSMode {
+func (o *grpcInboundOptions) effectiveMTLSMode(modes *grpcInboundModeStore)
func() grpcInboundMTLSMode {
if mode, ok := parseGRPCInboundMTLSMode(o.mtlsMode); ok {
- return mode
+ return func() grpcInboundMTLSMode { return mode }
}
- if mode, ok := grpcInboundMTLSModeFromRuntimeConfig(o.runtimeConfig,
upstreamPort(o.upstream)); ok {
- return mode
+ return modes.current
+}
+
+// grpcInboundModeStore caches the inbound mTLS mode read from the runtime
+// config. Reading it per connection would put a file read and a full JSON
+// parse on the accept path, and would let a transient read error silently
+// downgrade a STRICT port to PERMISSIVE.
+type grpcInboundModeStore struct {
+ path string
+ port int
+
+ mu sync.RWMutex
+ mode grpcInboundMTLSMode
+ loaded bool
+}
+
+func newGRPCInboundModeStore(path string, port int) *grpcInboundModeStore {
+ return &grpcInboundModeStore{path: path, port: port}
+}
+
+// current fails closed: until the runtime config has been read successfully
+// at least once, inbound traffic must present a client certificate.
+func (s *grpcInboundModeStore) current() grpcInboundMTLSMode {
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+ if !s.loaded {
+ return grpcInboundMTLSModeStrict
+ }
+ return s.mode
+}
+
+// reload replaces the cached mode only on a successful read and parse. Any
+// failure leaves the last known good mode in place, so a remount race or a
+// truncated write cannot relax the policy.
+func (s *grpcInboundModeStore) reload() error {
+ mode, err := loadGRPCInboundMTLSMode(s.path, s.port)
+ if err != nil {
+ return err
}
- return grpcInboundMTLSModePermissive
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.mode = mode
+ s.loaded = true
+ return nil
}
func parseGRPCInboundMTLSMode(mode string) (grpcInboundMTLSMode, bool) {
@@ -251,13 +520,21 @@ func upstreamPort(upstream string) int {
return out
}
-func grpcInboundMTLSModeFromRuntimeConfig(path string, port int)
(grpcInboundMTLSMode, bool) {
+// loadGRPCInboundMTLSMode reads the inbound mTLS mode for port from the
runtime
+// config. An absent config means the workload is unconfigured and yields
+// PERMISSIVE, matching a standalone run with no mounted secret. A config that
+// exists but cannot be read or parsed returns an error so the caller can keep
+// the last known mode instead of relaxing the policy.
+func loadGRPCInboundMTLSMode(path string, port int) (grpcInboundMTLSMode,
error) {
if path == "" {
- return "", false
+ return grpcInboundMTLSModePermissive, nil
}
data, err := os.ReadFile(path)
if err != nil {
- return "", false
+ if os.IsNotExist(err) {
+ return grpcInboundMTLSModePermissive, nil
+ }
+ return "", fmt.Errorf("read runtime config %s: %w", path, err)
}
var cfg struct {
Services []struct {
@@ -268,7 +545,7 @@ func grpcInboundMTLSModeFromRuntimeConfig(path string, port
int) (grpcInboundMTL
} `json:"services"`
}
if err := json.Unmarshal(data, &cfg); err != nil {
- return "", false
+ return "", fmt.Errorf("parse runtime config %s: %w", path, err)
}
foundDisable := false
@@ -283,19 +560,19 @@ func grpcInboundMTLSModeFromRuntimeConfig(path string,
port int) (grpcInboundMTL
continue
}
if mode == grpcInboundMTLSModeStrict {
- return grpcInboundMTLSModeStrict, true
+ return grpcInboundMTLSModeStrict, nil
}
foundPermissive = foundPermissive || mode ==
grpcInboundMTLSModePermissive
foundDisable = foundDisable || mode ==
grpcInboundMTLSModeDisable
}
}
if foundPermissive {
- return grpcInboundMTLSModePermissive, true
+ return grpcInboundMTLSModePermissive, nil
}
if foundDisable {
- return grpcInboundMTLSModeDisable, true
+ return grpcInboundMTLSModeDisable, nil
}
- return "", false
+ return grpcInboundMTLSModePermissive, nil
}
func copyBothDirections(a, b net.Conn) {
diff --git a/dubbod/discovery/cmd/app/grpc_inbound_test.go
b/dubbod/discovery/cmd/app/grpc_inbound_test.go
index adfc6433..25aac4b0 100644
--- a/dubbod/discovery/cmd/app/grpc_inbound_test.go
+++ b/dubbod/discovery/cmd/app/grpc_inbound_test.go
@@ -29,6 +29,7 @@ import (
"net"
"net/http"
"net/http/httptest"
+ neturl "net/url"
"os"
"path/filepath"
"testing"
@@ -48,7 +49,7 @@ func TestGRPCInboundRequiresClientCertificateAndProxiesHTTP(t
*testing.T) {
writePEM(t, filepath.Join(dir, "client-cert.pem"), "CERTIFICATE",
clientCert.Raw)
writePEM(t, filepath.Join(dir, "client-key.pem"), "RSA PRIVATE KEY",
x509.MarshalPKCS1PrivateKey(clientKey))
- tlsConfig, err :=
grpcInboundTLSConfigFromBootstrap(&xdsresolver.BootstrapConfig{
+ certs, err := newGRPCInboundCertStore(&xdsresolver.BootstrapConfig{
CertProviders: map[string]xdsresolver.FileWatcherCertConfig{
"default": {
CertificateFile: filepath.Join(dir,
"cert-chain.pem"),
@@ -58,8 +59,9 @@ func TestGRPCInboundRequiresClientCertificateAndProxiesHTTP(t
*testing.T) {
},
})
if err != nil {
- t.Fatalf("grpcInboundTLSConfigFromBootstrap() failed: %v", err)
+ t.Fatalf("newGRPCInboundCertStore() failed: %v", err)
}
+ tlsConfig := certs.tlsConfig(nil)
upstream := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, _ *http.Request) {
_, _ = fmt.Fprintln(w, "nginx v1")
@@ -136,7 +138,7 @@ func TestGRPCInboundPermissiveAcceptsPlaintextAndMTLS(t
*testing.T) {
writePEM(t, filepath.Join(dir, "client-cert.pem"), "CERTIFICATE",
clientCert.Raw)
writePEM(t, filepath.Join(dir, "client-key.pem"), "RSA PRIVATE KEY",
x509.MarshalPKCS1PrivateKey(clientKey))
- tlsConfig, err :=
grpcInboundTLSConfigFromBootstrap(&xdsresolver.BootstrapConfig{
+ certs, err := newGRPCInboundCertStore(&xdsresolver.BootstrapConfig{
CertProviders: map[string]xdsresolver.FileWatcherCertConfig{
"default": {
CertificateFile: filepath.Join(dir,
"cert-chain.pem"),
@@ -146,8 +148,9 @@ func TestGRPCInboundPermissiveAcceptsPlaintextAndMTLS(t
*testing.T) {
},
})
if err != nil {
- t.Fatalf("grpcInboundTLSConfigFromBootstrap() failed: %v", err)
+ t.Fatalf("newGRPCInboundCertStore() failed: %v", err)
}
+ tlsConfig := certs.tlsConfig(nil)
upstream := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, _ *http.Request) {
_, _ = fmt.Fprintln(w, "nginx v1")
@@ -223,11 +226,88 @@ func TestGRPCInboundMTLSModeFromRuntimeConfig(t
*testing.T) {
t.Fatalf("os.WriteFile() failed: %v", err)
}
- if got, ok := grpcInboundMTLSModeFromRuntimeConfig(path, 80); !ok ||
got != grpcInboundMTLSModePermissive {
- t.Fatalf("mode for 80 = %q, %v; want PERMISSIVE, true", got, ok)
+ if got, err := loadGRPCInboundMTLSMode(path, 80); err != nil || got !=
grpcInboundMTLSModePermissive {
+ t.Fatalf("mode for 80 = %q, %v; want PERMISSIVE, nil", got, err)
}
- if got, ok := grpcInboundMTLSModeFromRuntimeConfig(path, 8080); !ok ||
got != grpcInboundMTLSModeStrict {
- t.Fatalf("mode for 8080 = %q, %v; want STRICT, true", got, ok)
+ if got, err := loadGRPCInboundMTLSMode(path, 8080); err != nil || got
!= grpcInboundMTLSModeStrict {
+ t.Fatalf("mode for 8080 = %q, %v; want STRICT, nil", got, err)
+ }
+}
+
+func TestGRPCInboundModeStoreDoesNotDowngradeOnReadFailure(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "dubbo-grpc-xds.json")
+ if err := os.WriteFile(path,
[]byte(`{"services":[{"ports":[{"port":8080,"mtlsMode":"STRICT"}]}]}`), 0o600);
err != nil {
+ t.Fatalf("os.WriteFile() failed: %v", err)
+ }
+ store := newGRPCInboundModeStore(path, 8080)
+
+ if got := store.current(); got != grpcInboundMTLSModeStrict {
+ t.Fatalf("mode before first load = %q, want STRICT", got)
+ }
+ if err := store.reload(); err != nil {
+ t.Fatalf("reload() failed: %v", err)
+ }
+ if got := store.current(); got != grpcInboundMTLSModeStrict {
+ t.Fatalf("mode after load = %q, want STRICT", got)
+ }
+
+ // A truncated write must not relax the policy to PERMISSIVE.
+ if err := os.WriteFile(path, []byte(`{"services":`), 0o600); err != nil
{
+ t.Fatalf("os.WriteFile(truncated) failed: %v", err)
+ }
+ if err := store.reload(); err == nil {
+ t.Fatalf("reload() on malformed config returned nil error")
+ }
+ if got := store.current(); got != grpcInboundMTLSModeStrict {
+ t.Fatalf("mode after failed reload = %q, want STRICT", got)
+ }
+}
+
+func TestGRPCInboundModeStoreMissingConfigIsPermissive(t *testing.T) {
+ store := newGRPCInboundModeStore(filepath.Join(t.TempDir(),
"absent.json"), 8080)
+ if err := store.reload(); err != nil {
+ t.Fatalf("reload() on absent config failed: %v", err)
+ }
+ if got := store.current(); got != grpcInboundMTLSModePermissive {
+ t.Fatalf("mode for absent config = %q, want PERMISSIVE", got)
+ }
+}
+
+func TestGRPCInboundCertStoreReloadsRotatedCertificate(t *testing.T) {
+ caCert, caKey := newTestCA(t)
+ firstCert, firstKey := newSignedCert(t, caCert, caKey, "grpc-inbound")
+ dir := t.TempDir()
+ writePEM(t, filepath.Join(dir, "root-cert.pem"), "CERTIFICATE",
caCert.Raw)
+ writePEM(t, filepath.Join(dir, "cert-chain.pem"), "CERTIFICATE",
firstCert.Raw)
+ writePEM(t, filepath.Join(dir, "key.pem"), "RSA PRIVATE KEY",
x509.MarshalPKCS1PrivateKey(firstKey))
+
+ store, err := newGRPCInboundCertStore(&xdsresolver.BootstrapConfig{
+ CertProviders: map[string]xdsresolver.FileWatcherCertConfig{
+ "default": {
+ CertificateFile: filepath.Join(dir,
"cert-chain.pem"),
+ PrivateKeyFile: filepath.Join(dir,
"key.pem"),
+ CACertificateFile: filepath.Join(dir,
"root-cert.pem"),
+ },
+ },
+ })
+ if err != nil {
+ t.Fatalf("newGRPCInboundCertStore() failed: %v", err)
+ }
+ before, _ := store.current()
+
+ rotatedCert, rotatedKey := newSignedCert(t, caCert, caKey,
"grpc-inbound")
+ writePEM(t, filepath.Join(dir, "cert-chain.pem"), "CERTIFICATE",
rotatedCert.Raw)
+ writePEM(t, filepath.Join(dir, "key.pem"), "RSA PRIVATE KEY",
x509.MarshalPKCS1PrivateKey(rotatedKey))
+ if err := store.reload(); err != nil {
+ t.Fatalf("reload() failed: %v", err)
+ }
+
+ after, _ := store.current()
+ if string(before.Certificate[0]) == string(after.Certificate[0]) {
+ t.Fatalf("certificate was not reloaded after rotation")
+ }
+ if string(after.Certificate[0]) != string(rotatedCert.Raw) {
+ t.Fatalf("reloaded certificate does not match the rotated one")
}
}
@@ -298,3 +378,181 @@ func writePEM(t *testing.T, path, typ string, der []byte)
{
t.Fatalf("pem.Encode(%s) failed: %v", path, err)
}
}
+
+func TestGRPCInboundDefaultAcceptTimeoutIsSet(t *testing.T) {
+ t.Setenv("DUBBO_GRPC_INBOUND_ACCEPT_TIMEOUT", "")
+ cmd := newGRPCInboundCommand()
+ flag := cmd.Flags().Lookup("accept-timeout")
+ if flag == nil {
+ t.Fatalf("accept-timeout flag is missing")
+ }
+ if got, want := flag.DefValue, grpcInboundAcceptTimeout.String(); got
!= want {
+ t.Fatalf("accept-timeout default = %q, want %q", got, want)
+ }
+}
+
+func TestGRPCInboundPeerPolicy(t *testing.T) {
+ caCert, caKey := newTestCA(t)
+ local := newSPIFFECert(t, caCert, caKey,
"spiffe://cluster.local/ns/default/sa/reviews")
+ foreign := newSPIFFECert(t, caCert, caKey,
"spiffe://evil.example/ns/default/sa/reviews")
+ other := newSPIFFECert(t, caCert, caKey,
"spiffe://cluster.local/ns/default/sa/ratings")
+ noIdentity, _ := newSignedCert(t, caCert, caKey, "no-spiffe")
+
+ cases := []struct {
+ name string
+ policy *grpcInboundPeerPolicy
+ peer *x509.Certificate
+ wantErr bool
+ }{
+ {
+ name: "nil policy allows any peer",
+ policy: nil,
+ peer: foreign,
+ },
+ {
+ name: "empty policy allows any peer",
+ policy: &grpcInboundPeerPolicy{},
+ peer: foreign,
+ },
+ {
+ name: "matching trust domain is allowed",
+ policy: &grpcInboundPeerPolicy{trustDomain:
"cluster.local"},
+ peer: local,
+ },
+ {
+ name: "foreign trust domain is rejected",
+ policy: &grpcInboundPeerPolicy{trustDomain:
"cluster.local"},
+ peer: foreign,
+ wantErr: true,
+ },
+ {
+ name: "certificate without SPIFFE identity is
rejected",
+ policy: &grpcInboundPeerPolicy{trustDomain:
"cluster.local"},
+ peer: noIdentity,
+ wantErr: true,
+ },
+ {
+ name: "listed principal is allowed",
+ policy: &grpcInboundPeerPolicy{
+ trustDomain: "cluster.local",
+ allowed:
map[string]struct{}{"spiffe://cluster.local/ns/default/sa/reviews": {}},
+ },
+ peer: local,
+ },
+ {
+ name: "unlisted principal is rejected",
+ policy: &grpcInboundPeerPolicy{
+ trustDomain: "cluster.local",
+ allowed:
map[string]struct{}{"spiffe://cluster.local/ns/default/sa/reviews": {}},
+ },
+ peer: other,
+ wantErr: true,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ err := tc.policy.verifyPeerCertificate(nil,
[][]*x509.Certificate{{tc.peer}})
+ if tc.wantErr && err == nil {
+ t.Fatalf("verifyPeerCertificate() = nil, want
error")
+ }
+ if !tc.wantErr && err != nil {
+ t.Fatalf("verifyPeerCertificate() = %v, want
nil", err)
+ }
+ })
+ }
+}
+
+func TestGRPCInboundPeerPolicyRejectsEmptyChain(t *testing.T) {
+ policy := &grpcInboundPeerPolicy{trustDomain: "cluster.local"}
+ if err := policy.verifyPeerCertificate(nil, nil); err == nil {
+ t.Fatalf("verifyPeerCertificate() with no chain = nil, want
error")
+ }
+}
+
+func TestParseGRPCInboundPrincipals(t *testing.T) {
+ allowed, err := parseGRPCInboundPrincipals("ns/default/sa/reviews,
spiffe://other.mesh/ns/x/sa/y", "cluster.local")
+ if err != nil {
+ t.Fatalf("parseGRPCInboundPrincipals() failed: %v", err)
+ }
+ for _, want := range []string{
+ "spiffe://cluster.local/ns/default/sa/reviews",
+ "spiffe://other.mesh/ns/x/sa/y",
+ } {
+ if _, ok := allowed[want]; !ok {
+ t.Fatalf("principal %q missing from %v", want, allowed)
+ }
+ }
+
+ if got, err := parseGRPCInboundPrincipals("", ""); err != nil || got !=
nil {
+ t.Fatalf("parseGRPCInboundPrincipals(empty) = %v, %v; want nil,
nil", got, err)
+ }
+ if _, err := parseGRPCInboundPrincipals("ns/default/sa/reviews", "");
err == nil {
+ t.Fatalf("shorthand principal without trust domain returned nil
error")
+ }
+}
+
+func TestGRPCInboundCertStoreTrustDomain(t *testing.T) {
+ caCert, caKey := newTestCA(t)
+ leaf, key := newSPIFFECertWithKey(t, caCert, caKey,
"spiffe://cluster.local/ns/default/sa/reviews")
+ dir := t.TempDir()
+ writePEM(t, filepath.Join(dir, "root-cert.pem"), "CERTIFICATE",
caCert.Raw)
+ writePEM(t, filepath.Join(dir, "cert-chain.pem"), "CERTIFICATE",
leaf.Raw)
+ writePEM(t, filepath.Join(dir, "key.pem"), "RSA PRIVATE KEY",
x509.MarshalPKCS1PrivateKey(key))
+
+ store, err := newGRPCInboundCertStore(&xdsresolver.BootstrapConfig{
+ CertProviders: map[string]xdsresolver.FileWatcherCertConfig{
+ "default": {
+ CertificateFile: filepath.Join(dir,
"cert-chain.pem"),
+ PrivateKeyFile: filepath.Join(dir,
"key.pem"),
+ CACertificateFile: filepath.Join(dir,
"root-cert.pem"),
+ },
+ },
+ })
+ if err != nil {
+ t.Fatalf("newGRPCInboundCertStore() failed: %v", err)
+ }
+ if got, want := store.trustDomain(), "cluster.local"; got != want {
+ t.Fatalf("trustDomain() = %q, want %q", got, want)
+ }
+}
+
+func newSPIFFECert(t *testing.T, caCert *x509.Certificate, caKey
*rsa.PrivateKey, id string) *x509.Certificate {
+ t.Helper()
+ cert, _ := newSPIFFECertWithKey(t, caCert, caKey, id)
+ return cert
+}
+
+func newSPIFFECertWithKey(t *testing.T, caCert *x509.Certificate, caKey
*rsa.PrivateKey, id string) (*x509.Certificate, *rsa.PrivateKey) {
+ t.Helper()
+ uri, err := neturl.Parse(id)
+ if err != nil {
+ t.Fatalf("neturl.Parse(%s) failed: %v", id, err)
+ }
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatalf("rsa.GenerateKey() failed: %v", err)
+ }
+ serial, err := rand.Int(rand.Reader, big.NewInt(1<<62))
+ if err != nil {
+ t.Fatalf("rand.Int() failed: %v", err)
+ }
+ tmpl := &x509.Certificate{
+ SerialNumber: serial,
+ Subject: pkix.Name{CommonName: id},
+ NotBefore: time.Now().Add(-time.Minute),
+ NotAfter: time.Now().Add(time.Hour),
+ KeyUsage: x509.KeyUsageDigitalSignature |
x509.KeyUsageKeyEncipherment,
+ ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth,
x509.ExtKeyUsageClientAuth},
+ URIs: []*neturl.URL{uri},
+ }
+ der, err := x509.CreateCertificate(rand.Reader, tmpl, caCert,
&key.PublicKey, caKey)
+ if err != nil {
+ t.Fatalf("x509.CreateCertificate(%s) failed: %v", id, err)
+ }
+ cert, err := x509.ParseCertificate(der)
+ if err != nil {
+ t.Fatalf("x509.ParseCertificate(%s) failed: %v", id, err)
+ }
+ return cert, key
+}
