fix: Refresh client cert when CA rotation (#934)

When client CA is rotated, this can cause TLS read errors after the Dialer.Dial() has returned. The certificate should be refreshed.

This adds logic to the dialer to refresh the connection cache if a TLS error occurs after net.Conn is returned to the database driver.

Fixes #932

Author: hessjcg

dialer.go

@@ -419,9 +419,11 @@ func (d *Dialer) Dial(ctx context.Context, icn string, opts ...DialOption) (conn
 	tlsConn := tls.Client(conn, ci.TLSConfig())
 	err = tlsConn.HandshakeContext(ctx)
 	if err != nil {
+		// TLS handshake errors are fatal and require a refresh. Remove the instance
+		// from the cache so that future calls to Dial() will block until the
+		// certificate is refreshed successfully.
 		d.logger.Debugf(ctx, "[%v] TLS handshake failed: %v", cn.String(), err)
-		// refresh the instance info in case it caused the handshake failure
-		c.ForceRefresh()
+		d.removeCached(ctx, cn, c, err)
 		_ = tlsConn.Close() // best effort close attempt
 		return nil, errtype.NewDialError("handshake failed", cn.String(), err)
 	}
@@ -433,10 +435,26 @@ func (d *Dialer) Dial(ctx context.Context, icn string, opts ...DialOption) (conn
 		trace.RecordDialLatency(ctx, icn, d.dialerID, latency)
 	}()
 
-	iConn := newInstrumentedConn(tlsConn, func() {
+	closeFunc := func() {
 		n := atomic.AddUint64(c.openConnsCount, ^uint64(0)) // c.openConnsCount = c.openConnsCount - 1
 		trace.RecordOpenConnections(context.Background(), int64(n), d.dialerID, cn.String())
-	}, d.dialerID, cn.String())
+	}
+	errFunc := func(err error) {
+		// io.EOF occurs when the server closes the connection. This is safe to
+		// ignore.
+		if err == io.EOF {
+			return
+		}
+		d.logger.Debugf(ctx, "[%v] IO Error on Read or Write: %v", cn.String(), err)
+		if d.isTLSError(err) {
+			// TLS handshake errors are fatal. Remove the instance from the cache
+			// so that future calls to Dial() will block until the certificate
+			// is refreshed successfully.
+			d.removeCached(ctx, cn, c, err)
+			_ = tlsConn.Close() // best effort close attempt
+		}
+	}
+	iConn := newInstrumentedConn(tlsConn, closeFunc, errFunc, d.dialerID, cn.String())
 
 	// If this connection was opened using a Domain Name, then store it for later
 	// in case it needs to be forcibly closed.
@@ -447,23 +465,39 @@ func (d *Dialer) Dial(ctx context.Context, icn string, opts ...DialOption) (conn
 	}
 	return iConn, nil
 }
+func (d *Dialer) isTLSError(err error) bool {
+	if nErr, ok := err.(net.Error); ok {
+		return !nErr.Timeout() && // it's a permanent net error
+			strings.Contains(nErr.Error(), "tls") // it's a TLS-related error
+	}
+	return false
+}
 
-// removeCached stops all background refreshes and deletes the connection
-// info cache from the map of caches.
+// removeCached stops all background refreshes, closes open sockets, and deletes
+// the cache entry.
 func (d *Dialer) removeCached(
 	ctx context.Context,
-	i instance.ConnName, c connectionInfoCache, err error,
+	i instance.ConnName, c *monitoredCache, err error,
 ) {
 	d.logger.Debugf(
 		ctx,
 		"[%v] Removing connection info from cache: %v",
 		i.String(),
 		err,
 	)
+
+	// If this instance of monitoredCache is still in the cache, remove it.
+	// If this instance was already removed from the cache or
+	// if *a separate goroutine* replaced it with a new instance, do nothing.
+	key := createKey(i)
 	d.lock.Lock()
-	defer d.lock.Unlock()
+	if cachedC, ok := d.cache[key]; ok && cachedC == c {
+		delete(d.cache, key)
+	}
+	d.lock.Unlock()
+
+	// Close the monitoredCache, this call is idempotent.
 	c.Close()
-	delete(d.cache, createKey(i))
 }
 
 // validClientCert checks that the ephemeral client certificate retrieved from
@@ -505,7 +539,7 @@ func (d *Dialer) EngineVersion(ctx context.Context, icn string) (string, error)
 	}
 	ci, err := c.ConnectionInfo(ctx)
 	if err != nil {
-		d.removeCached(ctx, cn, c.connectionInfoCache, err)
+		d.removeCached(ctx, cn, c, err)
 		return "", err
 	}
 	return ci.DBVersion, nil
@@ -529,17 +563,18 @@ func (d *Dialer) Warmup(ctx context.Context, icn string, opts ...DialOption) err
 	}
 	_, err = c.ConnectionInfo(ctx)
 	if err != nil {
-		d.removeCached(ctx, cn, c.connectionInfoCache, err)
+		d.removeCached(ctx, cn, c, err)
 	}
 	return err
 }
 
 // newInstrumentedConn initializes an instrumentedConn that on closing will
 // decrement the number of open connects and record the result.
-func newInstrumentedConn(conn net.Conn, closeFunc func(), dialerID, connName string) *instrumentedConn {
+func newInstrumentedConn(conn net.Conn, closeFunc func(), errFunc func(error), dialerID, connName string) *instrumentedConn {
 	return &instrumentedConn{
 		Conn:      conn,
 		closeFunc: closeFunc,
+		errFunc:   errFunc,
 		dialerID:  dialerID,
 		connName:  connName,
 	}
@@ -550,6 +585,7 @@ func newInstrumentedConn(conn net.Conn, closeFunc func(), dialerID, connName str
 type instrumentedConn struct {
 	net.Conn
 	closeFunc func()
+	errFunc   func(error)
 	mu        sync.RWMutex
 	closed    bool
 	dialerID  string
@@ -562,6 +598,8 @@ func (i *instrumentedConn) Read(b []byte) (int, error) {
 	bytesRead, err := i.Conn.Read(b)
 	if err == nil {
 		go trace.RecordBytesReceived(context.Background(), int64(bytesRead), i.connName, i.dialerID)
+	} else {
+		i.errFunc(err)
 	}
 	return bytesRead, err
 }
@@ -572,6 +610,8 @@ func (i *instrumentedConn) Write(b []byte) (int, error) {
 	bytesWritten, err := i.Conn.Write(b)
 	if err == nil {
 		go trace.RecordBytesSent(context.Background(), int64(bytesWritten), i.connName, i.dialerID)
+	} else {
+		i.errFunc(err)
 	}
 	return bytesWritten, err
 }

dialer_test.go

@@ -1174,3 +1174,127 @@ func TestDialerChecksSubjectAlternativeNameAndFails(t *testing.T) {
 		t.Fatal("want error containing `tls: failed to verify certificate`. Got: ", err)
 	}
 }
+
+func TestDialerRefreshesAfterRotateCACerts(t *testing.T) {
+	tcs := []struct {
+		desc            string
+		rotateClientCA  bool
+		wantErrorOnDial bool
+		wantErrorOnRead bool
+		useLazyRefresh  bool
+	}{
+		{
+			desc:            "Rotating Client CA causes error on read, then refresh",
+			rotateClientCA:  true,
+			wantErrorOnRead: true,
+		},
+		{
+			desc:            "Rotating all CAs causes error on dial, then refresh",
+			wantErrorOnDial: true,
+			wantErrorOnRead: false,
+		},
+		{
+			desc:            "Rotating Client CA with lazy refresh causes error on read",
+			rotateClientCA:  true,
+			wantErrorOnRead: true,
+			useLazyRefresh:  true,
+		},
+		{
+			desc:            "Rotating all CAs with lazy refresh causes error on dial",
+			wantErrorOnDial: true,
+			useLazyRefresh:  true,
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.desc, func(t *testing.T) {
+			inst := mock.NewFakeCSQLInstanceWithSan(
+				"my-project", "my-region", "my-instance", []string{"db.example.com"},
+				mock.WithDNS("db.example.com"),
+				mock.WithServerCAMode("GOOGLE_MANAGED_CAS_CA"),
+			)
+
+			opts := []Option{
+				WithTokenSource(mock.EmptyTokenSource{}),
+				WithDebugLogger(&dialerTestLogger{t: t}),
+			}
+			if tc.useLazyRefresh {
+				opts = append(opts, WithLazyRefresh())
+			}
+
+			d := setupDialer(t, setupConfig{
+				skipServer:   true,
+				testInstance: inst,
+				reqs: []*mock.Request{
+					mock.InstanceGetSuccess(inst, 2),
+					mock.CreateEphemeralSuccess(inst, 2),
+				},
+				dialerOptions: opts,
+			})
+			cancel1 := mock.StartServerProxy(t, inst)
+			t.Log("First attempt...")
+			testSuccessfulDial(
+				context.Background(), t, d,
+				"my-project:my-region:my-instance",
+			)
+			t.Log("First attempt OK. Resetting client cert.")
+
+			// Close the server
+			cancel1()
+
+			if tc.rotateClientCA {
+				mock.RotateClientCA(inst)
+			} else {
+				mock.RotateCA(inst)
+			}
+
+			// Start the server with new certificates
+			cancel2 := mock.StartServerProxy(t, inst)
+			defer cancel2()
+
+			// Dial a second time.
+			t.Log("Second attempt should fail...")
+			conn, err := d.Dial(context.Background(), "my-project:my-region:my-instance")
+			if err != nil {
+				if tc.wantErrorOnDial {
+					t.Logf("got error on dial as expected: %v", err)
+				} else {
+					t.Fatalf("want no dial error, got: %v", err)
+				}
+			} else if tc.wantErrorOnDial {
+				t.Fatal("want dial error, got no error")
+			}
+
+			// If no error expected on dial, then attempt to read.
+			if !tc.wantErrorOnDial {
+				_, err = io.ReadAll(conn)
+				if err != nil {
+					if tc.wantErrorOnRead {
+						t.Logf("got error on read as expected: %v", err)
+					} else {
+						t.Fatalf("want no read error, got: %v", err)
+					}
+				} else if tc.wantErrorOnRead {
+					t.Fatal("want read error, got no error")
+				}
+			}
+			t.Log("Second attempt done")
+
+			// Dial again. This should complete after the refresh.
+			t.Log("Third attempt...")
+			testSuccessfulDial(
+				context.Background(), t, d,
+				"my-project:my-region:my-instance",
+			)
+			t.Log("Third attempt OK.")
+		})
+	}
+
+}
+
+type dialerTestLogger struct {
+	t *testing.T
+}
+
+func (l *dialerTestLogger) Debugf(f string, args ...interface{}) {
+	l.t.Logf(f, args...)
+}

internal/mock/sqladmin.go

@@ -102,38 +102,41 @@ func (r *Request) matches(hR *http.Request) bool {
 //
 // https://cloud.google.com/sql/docs/mysql/admin-api/rest/v1beta4/instances/get
 func InstanceGetSuccess(i FakeCSQLInstance, ct int) *Request {
-	var ips []*sqladmin.IpMapping
-	for ipType, addr := range i.ipAddrs {
-		if ipType == "PUBLIC" {
-			ips = append(ips, &sqladmin.IpMapping{IpAddress: addr, Type: "PRIMARY"})
-			continue
-		}
-		if ipType == "PRIVATE" {
-			ips = append(ips, &sqladmin.IpMapping{IpAddress: addr, Type: "PRIVATE"})
-		}
-	}
-
-	certBytes, err := i.serverCACert()
-	if err != nil {
-		panic(err)
-	}
-
-	db := &sqladmin.ConnectSettings{
-		BackendType:     i.backendType,
-		DatabaseVersion: i.dbVersion,
-		DnsName:         i.DNSName,
-		IpAddresses:     ips,
-		Region:          i.region,
-		ServerCaCert:    &sqladmin.SslCert{Cert: string(certBytes)},
-		PscEnabled:      i.pscEnabled,
-		ServerCaMode:    i.serverCAMode,
-	}
-
 	r := &Request{
 		reqMethod: http.MethodGet,
 		reqPath:   fmt.Sprintf("/sql/v1beta4/projects/%s/instances/%s/connectSettings", i.project, i.name),
 		reqCt:     ct,
 		handle: func(resp http.ResponseWriter, _ *http.Request) {
+			// Calculate the response when the request occurs the response contains
+			// up-to-date data stored in the FakeCSQLInstance.
+			// This is especially important for the i.serverCACert().
+			var ips []*sqladmin.IpMapping
+			for ipType, addr := range i.ipAddrs {
+				if ipType == "PUBLIC" {
+					ips = append(ips, &sqladmin.IpMapping{IpAddress: addr, Type: "PRIMARY"})
+					continue
+				}
+				if ipType == "PRIVATE" {
+					ips = append(ips, &sqladmin.IpMapping{IpAddress: addr, Type: "PRIVATE"})
+				}
+			}
+
+			certBytes, err := i.serverCACert()
+			if err != nil {
+				panic(err)
+			}
+
+			db := &sqladmin.ConnectSettings{
+				BackendType:     i.backendType,
+				DatabaseVersion: i.dbVersion,
+				DnsName:         i.DNSName,
+				IpAddresses:     ips,
+				Region:          i.region,
+				ServerCaCert:    &sqladmin.SslCert{Cert: string(certBytes)},
+				PscEnabled:      i.pscEnabled,
+				ServerCaMode:    i.serverCAMode,
+			}
+
 			b, err := db.MarshalJSON()
 			if err != nil {
 				http.Error(resp, err.Error(), http.StatusInternalServerError)