2017-05-11 21:38:54 +00:00
|
|
|
package cassandra
|
2017-04-13 20:48:32 +00:00
|
|
|
|
|
|
|
import (
|
|
|
|
"crypto/tls"
|
|
|
|
"fmt"
|
|
|
|
"strings"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/mitchellh/mapstructure"
|
|
|
|
|
|
|
|
"github.com/gocql/gocql"
|
|
|
|
"github.com/hashicorp/vault/helper/certutil"
|
2017-05-03 20:11:30 +00:00
|
|
|
"github.com/hashicorp/vault/helper/parseutil"
|
2017-04-13 20:48:32 +00:00
|
|
|
"github.com/hashicorp/vault/helper/tlsutil"
|
2017-05-11 21:38:54 +00:00
|
|
|
"github.com/hashicorp/vault/plugins/helper/database/connutil"
|
2017-04-13 20:48:32 +00:00
|
|
|
)
|
|
|
|
|
2017-05-11 21:38:54 +00:00
|
|
|
// cassandraConnectionProducer implements ConnectionProducer and provides an
|
2017-04-13 20:48:32 +00:00
|
|
|
// interface for cassandra databases to make connections.
|
2017-05-11 21:38:54 +00:00
|
|
|
type cassandraConnectionProducer struct {
|
2017-05-03 20:11:30 +00:00
|
|
|
Hosts string `json:"hosts" structs:"hosts" mapstructure:"hosts"`
|
|
|
|
Username string `json:"username" structs:"username" mapstructure:"username"`
|
|
|
|
Password string `json:"password" structs:"password" mapstructure:"password"`
|
|
|
|
TLS bool `json:"tls" structs:"tls" mapstructure:"tls"`
|
|
|
|
InsecureTLS bool `json:"insecure_tls" structs:"insecure_tls" mapstructure:"insecure_tls"`
|
|
|
|
ProtocolVersion int `json:"protocol_version" structs:"protocol_version" mapstructure:"protocol_version"`
|
|
|
|
ConnectTimeoutRaw interface{} `json:"connect_timeout" structs:"connect_timeout" mapstructure:"connect_timeout"`
|
|
|
|
TLSMinVersion string `json:"tls_min_version" structs:"tls_min_version" mapstructure:"tls_min_version"`
|
|
|
|
Consistency string `json:"consistency" structs:"consistency" mapstructure:"consistency"`
|
2017-05-03 22:36:49 +00:00
|
|
|
PemBundle string `json:"pem_bundle" structs:"pem_bundle" mapstructure:"pem_bundle"`
|
|
|
|
PemJSON string `json:"pem_json" structs:"pem_json" mapstructure:"pem_json"`
|
2017-05-03 20:11:30 +00:00
|
|
|
|
|
|
|
connectTimeout time.Duration
|
2017-05-03 22:36:49 +00:00
|
|
|
certificate string
|
|
|
|
privateKey string
|
|
|
|
issuingCA string
|
|
|
|
|
|
|
|
Initialized bool
|
|
|
|
Type string
|
|
|
|
session *gocql.Session
|
2017-04-13 20:48:32 +00:00
|
|
|
sync.Mutex
|
|
|
|
}
|
|
|
|
|
2017-05-11 21:38:54 +00:00
|
|
|
func (c *cassandraConnectionProducer) Initialize(conf map[string]interface{}, verifyConnection bool) error {
|
2017-04-13 20:48:32 +00:00
|
|
|
c.Lock()
|
|
|
|
defer c.Unlock()
|
|
|
|
|
2017-06-15 01:59:27 +00:00
|
|
|
err := mapstructure.WeakDecode(conf, c)
|
2017-04-13 20:48:32 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2017-05-03 20:45:27 +00:00
|
|
|
if c.ConnectTimeoutRaw == nil {
|
|
|
|
c.ConnectTimeoutRaw = "0s"
|
|
|
|
}
|
2017-05-03 20:11:30 +00:00
|
|
|
c.connectTimeout, err = parseutil.ParseDurationSecond(c.ConnectTimeoutRaw)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("invalid connect_timeout: %s", err)
|
|
|
|
}
|
|
|
|
|
2017-05-03 22:36:49 +00:00
|
|
|
switch {
|
|
|
|
case len(c.Hosts) == 0:
|
|
|
|
return fmt.Errorf("hosts cannot be empty")
|
|
|
|
case len(c.Username) == 0:
|
|
|
|
return fmt.Errorf("username cannot be empty")
|
|
|
|
case len(c.Password) == 0:
|
|
|
|
return fmt.Errorf("password cannot be empty")
|
|
|
|
}
|
|
|
|
|
|
|
|
var certBundle *certutil.CertBundle
|
|
|
|
var parsedCertBundle *certutil.ParsedCertBundle
|
|
|
|
switch {
|
|
|
|
case len(c.PemJSON) != 0:
|
|
|
|
parsedCertBundle, err = certutil.ParsePKIJSON([]byte(c.PemJSON))
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("could not parse given JSON; it must be in the format of the output of the PKI backend certificate issuing command: %s", err)
|
|
|
|
}
|
|
|
|
certBundle, err = parsedCertBundle.ToCertBundle()
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("Error marshaling PEM information: %s", err)
|
|
|
|
}
|
|
|
|
c.certificate = certBundle.Certificate
|
|
|
|
c.privateKey = certBundle.PrivateKey
|
|
|
|
c.issuingCA = certBundle.IssuingCA
|
|
|
|
c.TLS = true
|
|
|
|
|
|
|
|
case len(c.PemBundle) != 0:
|
|
|
|
parsedCertBundle, err = certutil.ParsePEMBundle(c.PemBundle)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("Error parsing the given PEM information: %s", err)
|
|
|
|
}
|
|
|
|
certBundle, err = parsedCertBundle.ToCertBundle()
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("Error marshaling PEM information: %s", err)
|
|
|
|
}
|
|
|
|
c.certificate = certBundle.Certificate
|
|
|
|
c.privateKey = certBundle.PrivateKey
|
|
|
|
c.issuingCA = certBundle.IssuingCA
|
|
|
|
c.TLS = true
|
|
|
|
}
|
|
|
|
|
2017-05-11 21:38:54 +00:00
|
|
|
// Set initialized to true at this point since all fields are set,
|
|
|
|
// and the connection can be established at a later time.
|
|
|
|
c.Initialized = true
|
|
|
|
|
2017-04-13 20:48:32 +00:00
|
|
|
if verifyConnection {
|
2017-04-23 01:02:57 +00:00
|
|
|
if _, err := c.Connection(); err != nil {
|
2017-05-11 21:38:54 +00:00
|
|
|
return fmt.Errorf("error verifying connection: %s", err)
|
2017-04-13 20:48:32 +00:00
|
|
|
}
|
|
|
|
}
|
2017-05-11 21:38:54 +00:00
|
|
|
|
2017-04-13 20:48:32 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2017-05-11 21:38:54 +00:00
|
|
|
func (c *cassandraConnectionProducer) Connection() (interface{}, error) {
|
2017-04-13 20:48:32 +00:00
|
|
|
if !c.Initialized {
|
2017-05-11 21:38:54 +00:00
|
|
|
return nil, connutil.ErrNotInitialized
|
2017-04-13 20:48:32 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// If we already have a DB, return it
|
|
|
|
if c.session != nil {
|
|
|
|
return c.session, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
session, err := c.createSession()
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Store the session in backend for reuse
|
|
|
|
c.session = session
|
|
|
|
|
|
|
|
return session, nil
|
|
|
|
}
|
|
|
|
|
2017-05-11 21:38:54 +00:00
|
|
|
func (c *cassandraConnectionProducer) Close() error {
|
2017-04-13 20:48:32 +00:00
|
|
|
// Grab the write lock
|
|
|
|
c.Lock()
|
|
|
|
defer c.Unlock()
|
|
|
|
|
|
|
|
if c.session != nil {
|
|
|
|
c.session.Close()
|
|
|
|
}
|
|
|
|
|
|
|
|
c.session = nil
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2017-05-11 21:38:54 +00:00
|
|
|
func (c *cassandraConnectionProducer) createSession() (*gocql.Session, error) {
|
2017-04-13 20:48:32 +00:00
|
|
|
clusterConfig := gocql.NewCluster(strings.Split(c.Hosts, ",")...)
|
|
|
|
clusterConfig.Authenticator = gocql.PasswordAuthenticator{
|
|
|
|
Username: c.Username,
|
|
|
|
Password: c.Password,
|
|
|
|
}
|
|
|
|
|
|
|
|
clusterConfig.ProtoVersion = c.ProtocolVersion
|
|
|
|
if clusterConfig.ProtoVersion == 0 {
|
|
|
|
clusterConfig.ProtoVersion = 2
|
|
|
|
}
|
|
|
|
|
2017-05-03 20:11:30 +00:00
|
|
|
clusterConfig.Timeout = c.connectTimeout
|
2017-04-13 20:48:32 +00:00
|
|
|
if c.TLS {
|
|
|
|
var tlsConfig *tls.Config
|
2017-05-03 22:36:49 +00:00
|
|
|
if len(c.certificate) > 0 || len(c.issuingCA) > 0 {
|
|
|
|
if len(c.certificate) > 0 && len(c.privateKey) == 0 {
|
2017-04-23 01:02:57 +00:00
|
|
|
return nil, fmt.Errorf("found certificate for TLS authentication but no private key")
|
2017-04-13 20:48:32 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
certBundle := &certutil.CertBundle{}
|
2017-05-03 22:36:49 +00:00
|
|
|
if len(c.certificate) > 0 {
|
|
|
|
certBundle.Certificate = c.certificate
|
|
|
|
certBundle.PrivateKey = c.privateKey
|
2017-04-13 20:48:32 +00:00
|
|
|
}
|
2017-05-03 22:36:49 +00:00
|
|
|
if len(c.issuingCA) > 0 {
|
|
|
|
certBundle.IssuingCA = c.issuingCA
|
2017-04-13 20:48:32 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
parsedCertBundle, err := certBundle.ToParsedCertBundle()
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("failed to parse certificate bundle: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
tlsConfig, err = parsedCertBundle.GetTLSConfig(certutil.TLSClient)
|
|
|
|
if err != nil || tlsConfig == nil {
|
|
|
|
return nil, fmt.Errorf("failed to get TLS configuration: tlsConfig:%#v err:%v", tlsConfig, err)
|
|
|
|
}
|
|
|
|
tlsConfig.InsecureSkipVerify = c.InsecureTLS
|
|
|
|
|
|
|
|
if c.TLSMinVersion != "" {
|
|
|
|
var ok bool
|
|
|
|
tlsConfig.MinVersion, ok = tlsutil.TLSLookup[c.TLSMinVersion]
|
|
|
|
if !ok {
|
|
|
|
return nil, fmt.Errorf("invalid 'tls_min_version' in config")
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// MinVersion was not being set earlier. Reset it to
|
|
|
|
// zero to gracefully handle upgrades.
|
|
|
|
tlsConfig.MinVersion = 0
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
clusterConfig.SslOpts = &gocql.SslOptions{
|
2017-04-19 18:19:29 +00:00
|
|
|
Config: tlsConfig,
|
2017-04-13 20:48:32 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
session, err := clusterConfig.CreateSession()
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("error creating session: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Set consistency
|
|
|
|
if c.Consistency != "" {
|
|
|
|
consistencyValue, err := gocql.ParseConsistencyWrapper(c.Consistency)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
session.SetConsistency(consistencyValue)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Verify the info
|
|
|
|
err = session.Query(`LIST USERS`).Exec()
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("error validating connection info: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
return session, nil
|
|
|
|
}
|