connecters with new config

This commit is contained in:
Anton Schirg 2018-08-27 15:19:17 +02:00
parent b955d308d9
commit 16e1396261

View File

@ -2,15 +2,14 @@ package cmd
import ( import (
"crypto/tls" "crypto/tls"
"fmt"
"net" "net"
"context" "context"
"github.com/jinzhu/copier" "github.com/jinzhu/copier"
"github.com/mitchellh/mapstructure"
"github.com/pkg/errors" "github.com/pkg/errors"
"github.com/problame/go-netssh" "github.com/problame/go-netssh"
"github.com/problame/go-streamrpc" "github.com/problame/go-streamrpc"
"github.com/zrepl/zrepl/cmd/config"
"github.com/zrepl/zrepl/cmd/tlsconf" "github.com/zrepl/zrepl/cmd/tlsconf"
"time" "time"
) )
@ -19,34 +18,26 @@ type SSHStdinserverConnecter struct {
Host string Host string
User string User string
Port uint16 Port uint16
IdentityFile string `mapstructure:"identity_file"` IdentityFile string
TransportOpenCommand []string `mapstructure:"transport_open_command"` TransportOpenCommand []string
SSHCommand string `mapstructure:"ssh_command"` SSHCommand string
Options []string Options []string
DialTimeout string `mapstructure:"dial_timeout"`
dialTimeout time.Duration dialTimeout time.Duration
} }
var _ streamrpc.Connecter = &SSHStdinserverConnecter{} var _ streamrpc.Connecter = &SSHStdinserverConnecter{}
func parseSSHStdinserverConnecter(i map[string]interface{}) (c *SSHStdinserverConnecter, err error) { func parseSSHStdinserverConnecter(in config.SSHStdinserverConnect) (c *SSHStdinserverConnecter, err error) {
c = &SSHStdinserverConnecter{} c = &SSHStdinserverConnecter{
if err = mapstructure.Decode(i, c); err != nil { Host: in.Host,
err = errors.New(fmt.Sprintf("could not parse ssh transport: %s", err)) User: in.User,
return nil, err Port: in.Port,
IdentityFile: in.IdentityFile,
SSHCommand: in.SSHCommand,
Options: in.Options,
dialTimeout: in.DialTimeout,
} }
if c.DialTimeout != "" {
c.dialTimeout, err = time.ParseDuration(c.DialTimeout)
if err != nil {
return nil, errors.Wrap(err, "cannot parse dial_timeout")
}
} else {
c.dialTimeout = 10 * time.Second
}
// TODO assert fields are filled
return return
} }
@ -78,77 +69,51 @@ func (c *SSHStdinserverConnecter) Connect(dialCtx context.Context) (net.Conn, er
} }
type TCPConnecter struct { type TCPConnecter struct {
Host string Address string
Port uint16 dialer net.Dialer
}
func parseTCPConnecter(in config.TCPConnect) (*TCPConnecter, error) {
dialer := net.Dialer{
Timeout: in.DialTimeout,
}
return &TCPConnecter{in.Address, dialer}, nil
}
func (c *TCPConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
return c.dialer.DialContext(dialCtx, "tcp", c.Address)
}
type TLSConnecter struct {
Address string
dialer net.Dialer dialer net.Dialer
tlsConfig *tls.Config tlsConfig *tls.Config
} }
func parseTCPConnecter(i map[string]interface{}) (*TCPConnecter, error) { func parseTLSConnecter(in config.TLSConnect) (*TLSConnecter, 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{ dialer := net.Dialer{
Timeout: dialTimeout, Timeout: in.DialTimeout,
} }
var tlsConfig *tls.Config ca, err := tlsconf.ParseCAFile(in.Ca)
if in.TLS != nil { if err != nil {
tlsConfig, err = func(i map[string]interface{}) (config *tls.Config, err error) { return nil, errors.Wrap(err, "cannot parse ca file")
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 cert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
if err != nil {
return nil, errors.Wrap(err, "cannot parse cert/key pair")
}
tlsConfig, err := tlsconf.ClientAuthClient(in.ServerCN, ca, cert)
if err != nil {
return nil, errors.Wrap(err, "cannot build tls config")
}
return &TLSConnecter{in.Address, dialer, tlsConfig}, nil
} }
func (c *TCPConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) { func (c *TLSConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
addr := fmt.Sprintf("%s:%d", c.Host, c.Port) return tls.DialWithDialer(&c.dialer, "tcp", c.Address, c.tlsConfig)
if c.tlsConfig != nil {
return tls.DialWithDialer(&c.dialer, "tcp", addr, c.tlsConfig)
}
return c.dialer.DialContext(dialCtx, "tcp", addr)
} }