*: refine options

Signed-off-by: xhe <xw897002528@gmail.com>

Author: xhebox

conf/weirproxy.yaml

@@ -20,30 +20,36 @@ log:
     max-backups: 1
 security:
   rsa-key-size: 4096
-  # tls object is either of type server or client
+  # tls object is either of type server, client, or peer
   # xxxx:
   #   ca: ca.pem
   #   cert: c.pem
   #   key: k.pem
   #   auto-certs: true
   #   skip-ca: trure
-  # client-tls:
+  # client object:
   #   1. requires: ca or skip-ca(skip verify server certs)
-  #   2. optionally: cert/key will be used for server-side client verification
+  #   2. optionally: cert/key will be used if server asks
   #   3. useless/forbid: auto-certs
   # server object:
   #   1. requires: cert/key or auto-certs(generate a temporary cert, mostly for testing)
-  #   2. optionally: ca will enable server-side client verification
+  #   2. optionally: ca will enable server-side client verification.
   #   3. useless/forbid: skip-ca
+  # peer object:
+  #   1. requires: cert/key/ca or auto-certs/skip-ca
   cluster-tls: # client object
     # access to other components like TiDB or PD, will use this
     skip-ca: true
   sql-tls: # client object
     # access to TiDB sql port, it has a standalone TLS configuration
     skip-ca: true
   server-tls: # server object
-    # proxy SQL or internal HTTP port will all use this
+    # proxy SQL or HTTP port will use this
     auto-certs: true
+  peer-tls: # peer object
+    # internal communication between proxies
+    auto-certs: true
+    skip-ca: true
 advance:
   # ignore-wrong-namespace: true
   # peer-port: "3081"

lib/config/proxy.go

@@ -94,6 +94,7 @@ func (c TLSConfig) HasCA() bool {
 type Security struct {
 	RSAKeySize int       `yaml:"rsa-key-size,omitempty" toml:"rsa-key-size,omitempty" json:"rsa-key-size,omitempty"`
 	ServerTLS  TLSConfig `yaml:"server-tls,omitempty" toml:"server-tls,omitempty" json:"server-tls,omitempty"`
+	PeerTLS    TLSConfig `yaml:"peer-tls,omitempty" toml:"peer-tls,omitempty" json:"peer-tls,omitempty"`
 	ClusterTLS TLSConfig `yaml:"cluster-tls,omitempty" toml:"cluster-tls,omitempty" json:"cluster-tls,omitempty"`
 	SQLTLS     TLSConfig `yaml:"sql-tls,omitempty" toml:"sql-tls,omitempty" json:"sql-tls,omitempty"`
 }

lib/config/proxy_test.go

@@ -63,6 +63,12 @@ var testProxyConfig = Config{
 			Key:       "c",
 			AutoCerts: true,
 		},
+		PeerTLS: TLSConfig{
+			CA:        "a",
+			Cert:      "b",
+			Key:       "c",
+			AutoCerts: true,
+		},
 		ClusterTLS: TLSConfig{
 			CA:     "a",
 			SkipCA: true,

lib/util/security/tls.go

@@ -22,7 +22,6 @@ import (
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/pem"
-	"fmt"
 	"io/ioutil"
 	"math/big"
 	"net"
@@ -45,6 +44,13 @@ func createTLSConfigificates(logger *zap.Logger, certpath string, keypath string
 		return errors.New("cert and key should be present or not at the same time")
 	}
 
+	if err := os.MkdirAll(filepath.Dir(keypath), 0755); err != nil {
+		return err
+	}
+	if err := os.MkdirAll(filepath.Dir(certpath), 0755); err != nil {
+		return err
+	}
+
 	privkey, err := rsa.GenerateKey(rand.Reader, rsaKeySize)
 	if err != nil {
 		return err
@@ -108,6 +114,18 @@ func createTLSConfigificates(logger *zap.Logger, certpath string, keypath string
 	return nil
 }
 
+func PreProcessTLSConfig(logger *zap.Logger, scfg *config.TLSConfig, workdir, mod string, keySize int) error {
+	if !scfg.HasCert() && scfg.AutoCerts {
+		scfg.Cert = filepath.Join(workdir, mod, "cert.pem")
+		scfg.Key = filepath.Join(workdir, mod, "key.pem")
+		if err := createTLSConfigificates(logger, scfg.Cert, scfg.Key, keySize); err != nil {
+			return errors.WithStack(err)
+		}
+		return PreProcessTLSConfig(logger, scfg, workdir, mod, keySize)
+	}
+	return nil
+}
+
 // CreateTLSConfigForTest is from https://gist.github.com/shaneutt/5e1995295cff6721c89a71d13a71c251.
 func CreateTLSConfigForTest() (serverTLSConf *tls.Config, clientTLSConf *tls.Config, err error) {
 	// set up our CA certificate
@@ -213,109 +231,98 @@ func CreateTLSConfigForTest() (serverTLSConf *tls.Config, clientTLSConf *tls.Con
 	return
 }
 
-func BuildServerTLSConfig(logger *zap.Logger, cfg config.TLSConfig, workdir, mod string, keySize int) (*tls.Config, error) {
+func BuildServerTLSConfig(logger *zap.Logger, cfg config.TLSConfig) (*tls.Config, error) {
+	logger = logger.With(zap.String("tls", "server"))
 	if !cfg.HasCert() {
-		if cfg.AutoCerts {
-			cfg.Cert = filepath.Join(workdir, mod, "cert.pem")
-			cfg.Key = filepath.Join(workdir, mod, "key.pem")
-			if err := createTLSConfigificates(logger, cfg.Cert, cfg.Key, keySize); err != nil {
-				return nil, err
-			}
-			return BuildServerTLSConfig(logger, cfg, workdir, mod, keySize)
-		}
-
-		// TODO: require certs here
-		logger.Warn(fmt.Sprintf("require certificates to secure %s clients connections", mod))
+		logger.Warn("require certificates to secure clients connections, disable TLS")
 		return nil, nil
 	}
 
 	tcfg := &tls.Config{}
 	cert, err := tls.LoadX509KeyPair(cfg.Cert, cfg.Key)
 	if err != nil {
-		return nil, errors.Errorf("failed to load certs for %s: %w", mod, err)
+		return nil, errors.Errorf("failed to load certs: %w", err)
 	}
 	tcfg.Certificates = append(tcfg.Certificates, cert)
 
 	if !cfg.HasCA() {
-		logger.Warn(fmt.Sprintf("no signed certs for %s, will not authenticate %s clients (connection is still secured)", mod, mod))
+		logger.Warn("no CA, server will not authenticate clients (connection is still secured)")
 		return tcfg, nil
 	}
 
 	tcfg.ClientAuth = tls.RequireAndVerifyClientCert
 	tcfg.ClientCAs = x509.NewCertPool()
 	certBytes, err := ioutil.ReadFile(cfg.CA)
 	if err != nil {
-		return nil, errors.Errorf("failed to read CA for %s: %w", mod, err)
+		return nil, errors.Errorf("failed to read CA: %w", err)
 	}
 	if !tcfg.ClientCAs.AppendCertsFromPEM(certBytes) {
-		return nil, errors.Errorf("failed to append CA for %s", mod)
+		return nil, errors.Errorf("failed to append CA")
 	}
 	return tcfg, nil
 }
 
-func BuildClientTLSConfig(logger *zap.Logger, cfg config.TLSConfig, mod string) (*tls.Config, error) {
+func BuildClientTLSConfig(logger *zap.Logger, cfg config.TLSConfig) (*tls.Config, error) {
+	logger = logger.With(zap.String("tls", "client"))
 	if !cfg.HasCA() {
-		logger.Warn(fmt.Sprintf("require CA to verify %s server connections", mod))
 		if cfg.SkipCA {
 			// still enable TLS without verify server certs
 			return &tls.Config{InsecureSkipVerify: true}, nil
 		}
-		// no TLS
+		logger.Warn("no CA to verify server connections, disable TLS")
 		return nil, nil
 	}
 
 	tcfg := &tls.Config{}
 	tcfg.ClientCAs = x509.NewCertPool()
 	certBytes, err := ioutil.ReadFile(cfg.CA)
 	if err != nil {
-		return nil, errors.Errorf("failed to read CA for %s: %w", mod, err)
+		return nil, errors.Errorf("failed to read CA: %w", err)
 	}
 	if !tcfg.ClientCAs.AppendCertsFromPEM(certBytes) {
-		return nil, errors.Errorf("failed to append CA for %s", mod)
+		return nil, errors.Errorf("failed to append CA")
 	}
 
 	if !cfg.HasCert() {
-		logger.Warn(fmt.Sprintf("no certs for %s, server may reject the connection", mod))
+		logger.Warn("no certificates, server may reject the connection")
 		return tcfg, nil
 	}
 	cert, err := tls.LoadX509KeyPair(cfg.Cert, cfg.Key)
 	if err != nil {
-		return nil, errors.Errorf("failed to load certs for %s: %w", mod, err)
+		return nil, errors.Errorf("failed to load certs for: %w", err)
 	}
 	tcfg.Certificates = append(tcfg.Certificates, cert)
 
 	return tcfg, nil
 }
 
-func BuildEtcdTLSConfig(logger *zap.Logger, server config.TLSConfig, workdir, mod string, keySize int) (clientInfo, peerInfo transport.TLSInfo, err error) {
-	if !server.HasCert() {
-		if server.AutoCerts {
-			server.Cert = filepath.Join(workdir, mod, "cert.pem")
-			server.Key = filepath.Join(workdir, mod, "key.pem")
-			if err = createTLSConfigificates(logger, server.Cert, server.Key, keySize); err != nil {
-				return
-			}
-			return BuildEtcdTLSConfig(logger, server, workdir, mod, keySize)
-		}
-	} else {
+func BuildEtcdTLSConfig(logger *zap.Logger, server, peer config.TLSConfig) (clientInfo, peerInfo transport.TLSInfo, err error) {
+	logger = logger.With(zap.String("tls", "etcd"))
+
+	if server.HasCert() {
 		clientInfo.CertFile = server.Cert
 		clientInfo.KeyFile = server.Key
 		if server.HasCA() {
 			clientInfo.ClientCertAuth = true
 			clientInfo.TrustedCAFile = server.CA
-		} else {
-			logger.Warn("no signed certs for etcd clients, proxy will not authenticate etcd clients (connection is still secured)")
+		} else if !server.SkipCA {
+			logger.Warn("no CA, proxy will not authenticate etcd clients (connection is still secured)")
 		}
 	}
 
-	if server.HasCA() && server.HasCert() {
-		peerInfo.CertFile = server.Cert
-		peerInfo.KeyFile = server.Key
-		peerInfo.TrustedCAFile = server.CA
-		peerInfo.ClientCertAuth = true
-	} else if server.HasCA() || server.HasCert() {
-		err = errors.New("need a full set of cert/ca/key for secure etcd peer inter-communication")
-		return
+	if peer.HasCert() {
+		peerInfo.CertFile = peer.Cert
+		peerInfo.KeyFile = peer.Key
+		if peer.HasCA() {
+			peerInfo.TrustedCAFile = peer.CA
+			peerInfo.ClientCertAuth = true
+		} else if peer.SkipCA {
+			peerInfo.InsecureSkipVerify = true
+			peerInfo.ClientCertAuth = false
+		} else {
+			err = errors.New("need a full set of cert/key/ca or cert/key/skip-ca for secure etcd peer inter-communication")
+			return
+		}
 	}
 
 	return

pkg/manager/config/manager.go

@@ -83,7 +83,7 @@ func (srv *ConfigManager) Init(ctx context.Context, addrs []string, cfg config.A
 	}
 
 	var err error
-	etcdConfig.TLS, err = security.BuildClientTLSConfig(logger, scfg, "frontend")
+	etcdConfig.TLS, err = security.BuildClientTLSConfig(logger, scfg)
 	if err != nil {
 		return errors.Wrapf(err, "create etcd config center error")
 	}

pkg/manager/namespace/manager.go

@@ -17,7 +17,7 @@ package namespace
 
 import (
 	"fmt"
-	"path/filepath"
+	"net/http"
 	"sync"
 
 	"github.com/pingcap/TiProxy/lib/config"
@@ -31,18 +31,18 @@ import (
 type NamespaceManager struct {
 	sync.RWMutex
 	client  *clientv3.Client
+	httpCli *http.Client
 	logger  *zap.Logger
-	workdir string
-	keySize int
 	nsm     map[string]*Namespace
 }
 
-func NewNamespaceManager(workdir string, keySize int) *NamespaceManager {
-	return &NamespaceManager{workdir: workdir, keySize: keySize}
+func NewNamespaceManager() *NamespaceManager {
+	return &NamespaceManager{}
 }
-func (mgr *NamespaceManager) buildNamespace(cfg *config.Namespace, client *clientv3.Client) (*Namespace, error) {
+func (mgr *NamespaceManager) buildNamespace(cfg *config.Namespace) (*Namespace, error) {
 	logger := mgr.logger.With(zap.String("namespace", cfg.Namespace))
-	rt, err := router.NewScoreBasedRouter(&cfg.Backend, client)
+
+	rt, err := router.NewScoreBasedRouter(&cfg.Backend, mgr.client, mgr.httpCli)
 	if err != nil {
 		return nil, errors.Errorf("build router error: %w", err)
 	}
@@ -51,12 +51,12 @@ func (mgr *NamespaceManager) buildNamespace(cfg *config.Namespace, client *clien
 		router: rt,
 	}
 
-	r.frontendTLS, err = security.BuildServerTLSConfig(logger, cfg.Frontend.Security, filepath.Join(mgr.workdir, r.name), "frontend", mgr.keySize)
+	r.frontendTLS, err = security.BuildServerTLSConfig(logger, cfg.Frontend.Security)
 	if err != nil {
 		return nil, errors.Errorf("build router error: %w", err)
 	}
 
-	r.backendTLS, err = security.BuildClientTLSConfig(logger, cfg.Backend.Security, "backend")
+	r.backendTLS, err = security.BuildClientTLSConfig(logger, cfg.Backend.Security)
 	if err != nil {
 		return nil, errors.Errorf("build router error: %w", err)
 	}
@@ -78,7 +78,7 @@ func (mgr *NamespaceManager) CommitNamespaces(nss []*config.Namespace, nss_delet
 			continue
 		}
 
-		ns, err := mgr.buildNamespace(nsc, mgr.client)
+		ns, err := mgr.buildNamespace(nsc)
 		if err != nil {
 			return fmt.Errorf("%w: create namespace error, namespace: %s", err, nsc.Namespace)
 		}
@@ -91,9 +91,10 @@ func (mgr *NamespaceManager) CommitNamespaces(nss []*config.Namespace, nss_delet
 	return nil
 }
 
-func (mgr *NamespaceManager) Init(logger *zap.Logger, nss []*config.Namespace, client *clientv3.Client) error {
+func (mgr *NamespaceManager) Init(logger *zap.Logger, nss []*config.Namespace, client *clientv3.Client, httpCli *http.Client) error {
 	mgr.Lock()
 	mgr.client = client
+	mgr.httpCli = httpCli
 	mgr.logger = logger
 	mgr.Unlock()
 

pkg/manager/router/backend_observer.go

@@ -132,6 +132,8 @@ type BackendObserver struct {
 	// All the backend info in the topology, including tombstones.
 	allBackendInfo map[string]*BackendInfo
 	client         *clientv3.Client
+	httpCli        *http.Client
+	httpTLS        bool
 	staticAddrs    []string
 	eventReceiver  BackendEventReceiver
 	wg             waitgroup.WaitGroup
@@ -149,7 +151,7 @@ func InitEtcdClient(logger *zap.Logger, cfg *config.Config) (*clientv3.Client, e
 	pdEndpoints := strings.Split(pdAddr, ",")
 	logConfig := zap.NewProductionConfig()
 	logConfig.Level = zap.NewAtomicLevelAt(zap.ErrorLevel)
-	tlsConfig, err := security.BuildClientTLSConfig(logger, cfg.Security.ClusterTLS, "pd")
+	tlsConfig, err := security.BuildClientTLSConfig(logger, cfg.Security.ClusterTLS)
 	if err != nil {
 		return nil, err
 	}
@@ -182,8 +184,8 @@ func InitEtcdClient(logger *zap.Logger, cfg *config.Config) (*clientv3.Client, e
 }
 
 // StartBackendObserver creates a BackendObserver and starts watching.
-func StartBackendObserver(eventReceiver BackendEventReceiver, client *clientv3.Client, config *HealthCheckConfig, staticAddrs []string) (*BackendObserver, error) {
-	bo, err := NewBackendObserver(eventReceiver, client, config, staticAddrs)
+func StartBackendObserver(eventReceiver BackendEventReceiver, client *clientv3.Client, httpCli *http.Client, config *HealthCheckConfig, staticAddrs []string) (*BackendObserver, error) {
+	bo, err := NewBackendObserver(eventReceiver, client, httpCli, config, staticAddrs)
 	if err != nil {
 		return nil, err
 	}
@@ -192,15 +194,24 @@ func StartBackendObserver(eventReceiver BackendEventReceiver, client *clientv3.C
 }
 
 // NewBackendObserver creates a BackendObserver.
-func NewBackendObserver(eventReceiver BackendEventReceiver, client *clientv3.Client, config *HealthCheckConfig, staticAddrs []string) (*BackendObserver, error) {
+func NewBackendObserver(eventReceiver BackendEventReceiver, client *clientv3.Client, httpCli *http.Client, config *HealthCheckConfig, staticAddrs []string) (*BackendObserver, error) {
 	if client == nil && len(staticAddrs) == 0 {
 		return nil, ErrNoInstanceToSelect
 	}
+	if httpCli == nil {
+		httpCli = http.DefaultClient
+	}
+	httpTLS := false
+	if v, ok := httpCli.Transport.(*http.Transport); ok && v != nil && v.TLSClientConfig != nil {
+		httpTLS = true
+	}
 	bo := &BackendObserver{
 		config:         config,
 		curBackendInfo: make(map[string]BackendStatus),
 		allBackendInfo: make(map[string]*BackendInfo),
 		client:         client,
+		httpCli:        httpCli,
+		httpTLS:        httpTLS,
 		staticAddrs:    staticAddrs,
 		eventReceiver:  eventReceiver,
 	}
@@ -382,11 +393,15 @@ func (bo *BackendObserver) checkHealth(ctx context.Context, backends map[string]
 
 		// When a backend gracefully shut down, the status port returns 500 but the SQL port still accepts
 		// new connections, so we must check the status port first.
-		url := fmt.Sprintf("http://%s:%d%s", info.IP, info.StatusPort, statusPathSuffix)
+		schema := "http"
+		if bo.httpTLS {
+			schema = "https"
+		}
+		url := fmt.Sprintf("%s://%s:%d%s", schema, info.IP, info.StatusPort, statusPathSuffix)
 		var resp *http.Response
 		err := connectWithRetry(func() error {
 			var err error
-			if resp, err = http.Get(url); err == nil {
+			if resp, err = bo.httpCli.Get(url); err == nil {
 				if err := resp.Body.Close(); err != nil {
 					logutil.Logger(ctx).Warn("close http response in health check failed", zap.Error(err))
 				}