implement tcp and tcp+tls transports

cmd/config_connect.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"crypto/tls"
 	"fmt"
 	"net"
 
@@ -11,6 +12,7 @@ import (
 	"github.com/problame/go-netssh"
 	"github.com/problame/go-streamrpc"
 	"time"
+	"github.com/zrepl/zrepl/cmd/tlsconf"
 )
 
 type SSHStdinserverConnecter struct {
@@ -74,3 +76,79 @@ func (c *SSHStdinserverConnecter) Connect(dialCtx context.Context) (net.Conn, er
 	}
 	return netsshConnToConn{nconn}, nil
 }
+
+type TCPConnecter struct {
+	Host      string
+	Port      uint16
+	dialer    net.Dialer
+	tlsConfig *tls.Config
+}
+
+func parseTCPConnecter(i map[string]interface{}) (*TCPConnecter, error) {
+	var in struct {
+		Host        string
+		Port        uint16
+		DialTimeout string `mapstructure:"dial_timeout"`
+		TLS         map[string]interface{}
+	}
+	if err := mapstructure.Decode(i, &in); err != nil {
+		return nil, errors.Wrap(err, "mapstructure error")
+	}
+
+	if in.Host == "" || in.Port == 0 {
+		return nil, errors.New("fields 'host' and 'port' must not be empty")
+	}
+	dialTimeout, err := parsePostitiveDuration(in.DialTimeout)
+	if err != nil {
+		if in.DialTimeout != "" {
+			return nil, errors.Wrap(err, "cannot parse field 'dial_timeout'")
+		}
+		dialTimeout = 10 * time.Second
+	}
+	dialer := net.Dialer{
+		Timeout: dialTimeout,
+	}
+
+	var tlsConfig *tls.Config
+	if in.TLS != nil {
+		tlsConfig, err = func(i map[string]interface{}) (config *tls.Config, err error) {
+			var in struct {
+				CA       string
+				Cert     string
+				Key      string
+				ServerCN string `mapstructure:"server_cn"`
+			}
+			if err := mapstructure.Decode(i, &in); err != nil {
+				return nil, errors.Wrap(err, "mapstructure error")
+			}
+			if in.CA == "" || in.Cert == "" || in.Key == "" || in.ServerCN == "" {
+				return nil, errors.New("fields 'ca', 'cert', 'key' and 'server_cn' must be specified")
+			}
+
+			ca, err := tlsconf.ParseCAFile(in.CA)
+			if err != nil {
+				return nil, errors.Wrap(err, "cannot parse ca file")
+			}
+
+			cert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
+			if err != nil {
+				return nil, errors.Wrap(err, "cannot parse cert/key pair")
+			}
+
+			return tlsconf.ClientAuthClient(in.ServerCN, ca, cert)
+		}(in.TLS)
+		if err != nil {
+			return nil, errors.Wrap(err, "cannot parse TLS config in field 'tls'")
+		}
+	}
+
+	return &TCPConnecter{in.Host, in.Port, dialer, tlsConfig}, nil
+}
+
+func (c *TCPConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
+	addr := fmt.Sprintf("%s:%d", c.Host, c.Port)
+	if c.tlsConfig != nil {
+		return tls.DialWithDialer(&c.dialer, "tcp", addr, c.tlsConfig)
+	}
+	return c.dialer.DialContext(dialCtx, "tcp", addr)
+}

cmd/config_job_pull.go

@@ -51,7 +51,7 @@ func parsePullJob(c JobParsingContext, name string, i map[string]interface{}) (j
 
 	j = &PullJob{Name: name}
 
-	j.Connect, err = parseSSHStdinserverConnecter(asMap.Connect)
+	j.Connect, err = parseConnect(asMap.Connect)
 	if err != nil {
 		err = errors.Wrap(err, "cannot parse 'connect'")
 		return nil, err
@@ -167,10 +167,7 @@ func (j *PullJob) doRun(ctx context.Context) {
 		ConnConfig: STREAMRPC_CONFIG,
 	}
 
-	//client, err := streamrpc.NewClient(j.Connect, clientConf)
-	client, err := streamrpc.NewClient(&tcpConnecter{net.Dialer{
-		Timeout: 10*time.Second,	
-	}}, clientConf)
+	client, err := streamrpc.NewClient(j.Connect, clientConf)
 	defer client.Close()
 
 	j.task.Enter("pull")

cmd/config_job_source.go

@@ -147,9 +147,7 @@ func (j *SourceJob) Pruner(task *Task, side PrunePolicySide, dryRun bool) (p Pru
 
 func (j *SourceJob) serve(ctx context.Context, task *Task) {
 
-	//listener, err := j.Serve.Listen()
-
-	listener, err := net.Listen("tcp", ":8888")
+	listener, err := j.Serve.Listen()
 	if err != nil {
 		task.Log().WithError(err).Error("error listening")
 		return

cmd/config_logging.go

@@ -1,15 +1,15 @@
 package cmd
 
 import (
-	"crypto/tls"
-	"crypto/x509"
 	"github.com/mattn/go-isatty"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 	"github.com/zrepl/zrepl/logger"
-	"io/ioutil"
 	"os"
 	"time"
+	"crypto/tls"
+	"crypto/x509"
+	"github.com/zrepl/zrepl/cmd/tlsconf"
 )
 
 type LoggingConfig struct {
@@ -164,11 +164,7 @@ func parseTCPOutlet(i interface{}, formatter EntryFormatter) (out *TCPOutlet, er
 		Net           string
 		Address       string
 		RetryInterval string `mapstructure:"retry_interval"`
-		TLS           *struct {
-			CA   string
-			Cert string
-			Key  string
-		}
+		TLS           map[string]interface{}
 	}
 	if err = mapstructure.Decode(i, &in); err != nil {
 		return nil, errors.Wrap(err, "mapstructure error")
@@ -188,37 +184,41 @@ func parseTCPOutlet(i interface{}, formatter EntryFormatter) (out *TCPOutlet, er
 
 	var tlsConfig *tls.Config
 	if in.TLS != nil {
-
-		cert, err := tls.LoadX509KeyPair(in.TLS.Cert, in.TLS.Key)
-		if err != nil {
-			return nil, errors.Wrap(err, "cannot load client cert")
-		}
-
-		var rootCAs *x509.CertPool
-		if in.TLS.CA == "" {
-			if rootCAs, err = x509.SystemCertPool(); err != nil {
-				return nil, errors.Wrap(err, "cannot open system cert pool")
+		tlsConfig, err = func(m map[string]interface{}, host string) (*tls.Config, error) {
+			var in struct {
+				CA   string
+				Cert string
+				Key  string
 			}
-		} else {
-			rootCAs = x509.NewCertPool()
-			rootCAPEM, err := ioutil.ReadFile(in.TLS.CA)
+			if err := mapstructure.Decode(m, &in); err != nil {
+				return nil, errors.Wrap(err, "mapstructure error")
+			}
+
+			clientCert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
 			if err != nil {
-				return nil, errors.Wrap(err, "cannot load CA cert")
+				return nil, errors.Wrap(err, "cannot load client cert")
 			}
-			if !rootCAs.AppendCertsFromPEM(rootCAPEM) {
-				return nil, errors.New("cannot parse CA cert")
+
+			var rootCAs *x509.CertPool
+			if in.CA == "" {
+				if rootCAs, err = x509.SystemCertPool(); err != nil {
+					return nil, errors.Wrap(err, "cannot open system cert pool")
+				}
+			} else {
+				rootCAs, err = tlsconf.ParseCAFile(in.CA)
+				if err != nil {
+					return nil, errors.Wrap(err, "cannot parse CA cert")
+				}
+			}
+			if rootCAs == nil {
+				panic("invariant violated")
 			}
-		}
-		if err != nil && in.TLS.CA == "" {
-			return nil, errors.Wrap(err, "cannot load root ca pool")
-		}
 
-		tlsConfig = &tls.Config{
-			Certificates: []tls.Certificate{cert},
-			RootCAs:      rootCAs,
+			return tlsconf.ClientAuthClient(host, rootCAs, clientCert)
+		}(in.TLS, in.Address)
+		if err != nil {
+			return nil, errors.New("cannot not parse TLS config in field 'tls'")
 		}
-
-		tlsConfig.BuildNameToCertificate()
 	}
 
 	formatter.SetMetadataFlags(MetadataAll)

cmd/config_parse.go

@@ -220,6 +220,8 @@ func parseConnect(i map[string]interface{}) (c streamrpc.Connecter, err error) {
 	switch t {
 	case "ssh+stdinserver":
 		return parseSSHStdinserverConnecter(i)
+	case "tcp":
+		return parseTCPConnecter(i)
 	default:
 		return nil, errors.Errorf("unknown connection type '%s'", t)
 	}
@@ -278,6 +280,8 @@ func parseAuthenticatedChannelListenerFactory(c JobParsingContext, v map[string]
 	switch t {
 	case "stdinserver":
 		return parseStdinserverListenerFactory(c, v)
+	case "tcp":
+		return parseTCPListenerFactory(c, v)
 	default:
 		err = errors.Errorf("unknown type '%s'", t)
 		return

cmd/config_serve_tcp.go

@@ -0,0 +1,86 @@
+package cmd
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"net"
+	"time"
+
+	"github.com/mitchellh/mapstructure"
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/cmd/tlsconf"
+)
+
+type TCPListenerFactory struct {
+	Address          string
+	tls              bool
+	clientCA         *x509.CertPool
+	serverCert       tls.Certificate
+	clientCommonName string
+}
+
+func parseTCPListenerFactory(c JobParsingContext, i map[string]interface{}) (*TCPListenerFactory, error) {
+
+	var in struct {
+		Address        string
+		TLS            map[string]interface{}
+	}
+	if err := mapstructure.Decode(i, &in); err != nil {
+		return nil, errors.Wrap(err, "mapstructure error")
+	}
+
+	lf := &TCPListenerFactory{}
+
+	if in.Address == "" {
+		return nil, errors.New("must specify field 'address'")
+	}
+	lf.Address = in.Address
+
+	if in.TLS != nil {
+		err := func(i map[string]interface{}) (err error) {
+			var in struct {
+				CA       string
+				Cert     string
+				Key      string
+				ClientCN string `mapstructure:"client_cn"`
+			}
+			if err := mapstructure.Decode(i, &in); err != nil {
+				return errors.Wrap(err, "mapstructure error")
+			}
+
+			if in.CA == "" || in.Cert == "" || in.Key == "" || in.ClientCN == "" {
+				return errors.New("fields 'ca', 'cert', 'key' and 'client_cn' must be specified")
+			}
+
+			lf.clientCommonName = in.ClientCN
+
+			lf.clientCA, err = tlsconf.ParseCAFile(in.CA)
+			if err != nil {
+				return errors.Wrap(err,"cannot parse ca file")
+			}
+
+			lf.serverCert, err = tls.LoadX509KeyPair(in.Cert, in.Key)
+			if err != nil {
+				return errors.Wrap(err, "cannot parse cer/key pair")
+			}
+
+			lf.tls = true // mark success
+			return nil
+		}(in.TLS)
+		if err != nil {
+			return nil, errors.Wrap(err, "error parsing TLS config in field 'tls'")
+		}
+	}
+
+	return lf, nil
+}
+
+func (f *TCPListenerFactory) Listen() (net.Listener, error) {
+	l, err := net.Listen("tcp", f.Address)
+	if !f.tls || err != nil {
+		return l, err
+	}
+
+	tl := tlsconf.NewClientAuthListener(l, f.clientCA, f.serverCert, f.clientCommonName, 10*time.Second)
+	return tl, nil
+}

cmd/tlsconf/tlsconf.go

@@ -0,0 +1,121 @@
+package tlsconf
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"time"
+)
+
+func ParseCAFile(certfile string) (*x509.CertPool, error) {
+	pool := x509.NewCertPool()
+	pem, err := ioutil.ReadFile(certfile)
+	if err != nil {
+		return nil, err
+	}
+	if !pool.AppendCertsFromPEM(pem) {
+		return nil, errors.New("PEM parsing error")
+	}
+	return pool, nil
+}
+
+type ClientAuthListener struct {
+	l                net.Listener
+	clientCommonName string
+	handshakeTimeout time.Duration
+}
+
+func NewClientAuthListener(
+	l net.Listener, ca *x509.CertPool, serverCert tls.Certificate,
+	clientCommonName string, handshakeTimeout time.Duration) *ClientAuthListener {
+
+	if ca == nil {
+		panic(ca)
+	}
+	if serverCert.Certificate == nil || serverCert.PrivateKey == nil {
+		panic(serverCert)
+	}
+	if clientCommonName == "" {
+		panic(clientCommonName)
+	}
+
+	tlsConf := tls.Config{
+		Certificates: []tls.Certificate{serverCert},
+		ClientCAs:                ca,
+		ClientAuth:               tls.RequireAndVerifyClientCert,
+		PreferServerCipherSuites: true,
+	}
+	l = tls.NewListener(l, &tlsConf)
+	return &ClientAuthListener{
+		l,
+		clientCommonName,
+		handshakeTimeout,
+	}
+}
+
+func (l *ClientAuthListener) Accept() (c net.Conn, err error) {
+	c, err = l.l.Accept()
+	if err != nil {
+		return nil, err
+	}
+	tlsConn, ok := c.(*tls.Conn)
+	if !ok {
+		return c, err
+	}
+
+	var (
+		cn        string
+		peerCerts []*x509.Certificate
+	)
+	if err = tlsConn.SetDeadline(time.Now().Add(l.handshakeTimeout)); err != nil {
+		goto CloseAndErr
+	}
+	if err = tlsConn.Handshake(); err != nil {
+		goto CloseAndErr
+	}
+
+	peerCerts = tlsConn.ConnectionState().PeerCertificates
+	if len(peerCerts) != 1 {
+		err = errors.New("unexpected number of certificates presented by TLS client")
+		goto CloseAndErr
+	}
+	cn = peerCerts[0].Subject.CommonName
+	if cn != l.clientCommonName {
+		err = fmt.Errorf("client cert common name does not match client_identity: %q != %q", cn, l.clientCommonName)
+		goto CloseAndErr
+	}
+	return c, nil
+CloseAndErr:
+	c.Close()
+	return nil, err
+}
+
+func (l *ClientAuthListener) Addr() net.Addr {
+	return l.l.Addr()
+}
+
+func (l *ClientAuthListener) Close() error {
+	return l.l.Close()
+}
+
+func ClientAuthClient(serverName string, rootCA *x509.CertPool, clientCert tls.Certificate) (*tls.Config, error) {
+	if serverName == "" {
+		panic(serverName)
+	}
+	if rootCA == nil {
+		panic(rootCA)
+	}
+	if clientCert.Certificate == nil || clientCert.PrivateKey == nil {
+		panic(clientCert)
+	}
+	tlsConfig := &tls.Config{
+		Certificates: []tls.Certificate{clientCert},
+		RootCAs:      rootCA,
+		ServerName: serverName,
+	}
+	tlsConfig.BuildNameToCertificate()
+	return tlsConfig, nil
+}