245 lines
7.3 KiB
Go
245 lines
7.3 KiB
Go
// Copyright (c) HashiCorp, Inc.
|
|
// SPDX-License-Identifier: MPL-2.0
|
|
|
|
package cassandra
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/gocql/gocql"
|
|
"github.com/hashicorp/go-secure-stdlib/parseutil"
|
|
"github.com/hashicorp/go-secure-stdlib/tlsutil"
|
|
dbplugin "github.com/hashicorp/vault/sdk/database/dbplugin/v5"
|
|
"github.com/hashicorp/vault/sdk/database/helper/connutil"
|
|
"github.com/hashicorp/vault/sdk/database/helper/dbutil"
|
|
"github.com/mitchellh/mapstructure"
|
|
)
|
|
|
|
// cassandraConnectionProducer implements ConnectionProducer and provides an
|
|
// interface for cassandra databases to make connections.
|
|
type cassandraConnectionProducer struct {
|
|
Hosts string `json:"hosts" structs:"hosts" mapstructure:"hosts"`
|
|
Port int `json:"port" structs:"port" mapstructure:"port"`
|
|
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"`
|
|
TLSServerName string `json:"tls_server_name" structs:"tls_server_name" mapstructure:"tls_server_name"`
|
|
ProtocolVersion int `json:"protocol_version" structs:"protocol_version" mapstructure:"protocol_version"`
|
|
ConnectTimeoutRaw interface{} `json:"connect_timeout" structs:"connect_timeout" mapstructure:"connect_timeout"`
|
|
SocketKeepAliveRaw interface{} `json:"socket_keep_alive" structs:"socket_keep_alive" mapstructure:"socket_keep_alive"`
|
|
TLSMinVersion string `json:"tls_min_version" structs:"tls_min_version" mapstructure:"tls_min_version"`
|
|
Consistency string `json:"consistency" structs:"consistency" mapstructure:"consistency"`
|
|
LocalDatacenter string `json:"local_datacenter" structs:"local_datacenter" mapstructure:"local_datacenter"`
|
|
PemBundle string `json:"pem_bundle" structs:"pem_bundle" mapstructure:"pem_bundle"`
|
|
PemJSON string `json:"pem_json" structs:"pem_json" mapstructure:"pem_json"`
|
|
SkipVerification bool `json:"skip_verification" structs:"skip_verification" mapstructure:"skip_verification"`
|
|
|
|
connectTimeout time.Duration
|
|
socketKeepAlive time.Duration
|
|
sslOpts *gocql.SslOptions
|
|
rawConfig map[string]interface{}
|
|
|
|
Initialized bool
|
|
Type string
|
|
session *gocql.Session
|
|
sync.Mutex
|
|
}
|
|
|
|
func (c *cassandraConnectionProducer) Initialize(ctx context.Context, req dbplugin.InitializeRequest) error {
|
|
c.Lock()
|
|
defer c.Unlock()
|
|
|
|
c.rawConfig = req.Config
|
|
|
|
err := mapstructure.WeakDecode(req.Config, c)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if c.ConnectTimeoutRaw == nil {
|
|
c.ConnectTimeoutRaw = "5s"
|
|
}
|
|
c.connectTimeout, err = parseutil.ParseDurationSecond(c.ConnectTimeoutRaw)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid connect_timeout: %w", err)
|
|
}
|
|
|
|
if c.SocketKeepAliveRaw == nil {
|
|
c.SocketKeepAliveRaw = "0s"
|
|
}
|
|
c.socketKeepAlive, err = parseutil.ParseDurationSecond(c.SocketKeepAliveRaw)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid socket_keep_alive: %w", err)
|
|
}
|
|
|
|
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")
|
|
case len(c.PemJSON) > 0 && len(c.PemBundle) > 0:
|
|
return fmt.Errorf("cannot specify both pem_json and pem_bundle")
|
|
}
|
|
|
|
var tlsMinVersion uint16 = tls.VersionTLS12
|
|
if c.TLSMinVersion != "" {
|
|
ver, exists := tlsutil.TLSLookup[c.TLSMinVersion]
|
|
if !exists {
|
|
return fmt.Errorf("unrecognized TLS version [%s]", c.TLSMinVersion)
|
|
}
|
|
tlsMinVersion = ver
|
|
}
|
|
|
|
switch {
|
|
case len(c.PemJSON) != 0:
|
|
cfg, err := jsonBundleToTLSConfig(c.PemJSON, tlsMinVersion, c.TLSServerName, c.InsecureTLS)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to parse pem_json: %w", err)
|
|
}
|
|
c.sslOpts = &gocql.SslOptions{
|
|
Config: cfg,
|
|
EnableHostVerification: !cfg.InsecureSkipVerify,
|
|
}
|
|
c.TLS = true
|
|
|
|
case len(c.PemBundle) != 0:
|
|
cfg, err := pemBundleToTLSConfig(c.PemBundle, tlsMinVersion, c.TLSServerName, c.InsecureTLS)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to parse pem_bundle: %w", err)
|
|
}
|
|
c.sslOpts = &gocql.SslOptions{
|
|
Config: cfg,
|
|
EnableHostVerification: !cfg.InsecureSkipVerify,
|
|
}
|
|
c.TLS = true
|
|
|
|
case c.InsecureTLS:
|
|
c.sslOpts = &gocql.SslOptions{
|
|
EnableHostVerification: !c.InsecureTLS,
|
|
}
|
|
}
|
|
|
|
// 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
|
|
|
|
if req.VerifyConnection {
|
|
if _, err := c.Connection(ctx); err != nil {
|
|
return fmt.Errorf("error verifying connection: %w", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *cassandraConnectionProducer) Connection(ctx context.Context) (interface{}, error) {
|
|
if !c.Initialized {
|
|
return nil, connutil.ErrNotInitialized
|
|
}
|
|
|
|
// If we already have a DB, return it
|
|
if c.session != nil && !c.session.Closed() {
|
|
return c.session, nil
|
|
}
|
|
|
|
session, err := c.createSession(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Store the session in backend for reuse
|
|
c.session = session
|
|
|
|
return session, nil
|
|
}
|
|
|
|
func (c *cassandraConnectionProducer) Close() error {
|
|
c.Lock()
|
|
defer c.Unlock()
|
|
|
|
if c.session != nil {
|
|
c.session.Close()
|
|
}
|
|
|
|
c.session = nil
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *cassandraConnectionProducer) createSession(ctx context.Context) (*gocql.Session, error) {
|
|
hosts := strings.Split(c.Hosts, ",")
|
|
clusterConfig := gocql.NewCluster(hosts...)
|
|
clusterConfig.Authenticator = gocql.PasswordAuthenticator{
|
|
Username: c.Username,
|
|
Password: c.Password,
|
|
}
|
|
|
|
if c.Port != 0 {
|
|
clusterConfig.Port = c.Port
|
|
}
|
|
|
|
clusterConfig.ProtoVersion = c.ProtocolVersion
|
|
if clusterConfig.ProtoVersion == 0 {
|
|
clusterConfig.ProtoVersion = 2
|
|
}
|
|
|
|
clusterConfig.Timeout = c.connectTimeout
|
|
clusterConfig.ConnectTimeout = c.connectTimeout
|
|
clusterConfig.SocketKeepalive = c.socketKeepAlive
|
|
clusterConfig.SslOpts = c.sslOpts
|
|
|
|
if c.LocalDatacenter != "" {
|
|
clusterConfig.PoolConfig.HostSelectionPolicy = gocql.DCAwareRoundRobinPolicy(c.LocalDatacenter)
|
|
}
|
|
|
|
session, err := clusterConfig.CreateSession()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error creating session: %w", err)
|
|
}
|
|
|
|
if c.Consistency != "" {
|
|
consistencyValue, err := gocql.ParseConsistencyWrapper(c.Consistency)
|
|
if err != nil {
|
|
session.Close()
|
|
return nil, err
|
|
}
|
|
|
|
session.SetConsistency(consistencyValue)
|
|
}
|
|
|
|
if !c.SkipVerification {
|
|
err = session.Query(`LIST ALL`).WithContext(ctx).Exec()
|
|
if err != nil && len(c.Username) != 0 && strings.Contains(err.Error(), "not authorized") {
|
|
rowNum := session.Query(dbutil.QueryHelper(`LIST CREATE ON ALL ROLES OF '{{username}}';`, map[string]string{
|
|
"username": c.Username,
|
|
})).Iter().NumRows()
|
|
|
|
if rowNum < 1 {
|
|
session.Close()
|
|
return nil, fmt.Errorf("error validating connection info: No role create permissions found, previous error: %w", err)
|
|
}
|
|
} else if err != nil {
|
|
session.Close()
|
|
return nil, fmt.Errorf("error validating connection info: %w", err)
|
|
}
|
|
}
|
|
|
|
return session, nil
|
|
}
|
|
|
|
func (c *cassandraConnectionProducer) secretValues() map[string]string {
|
|
return map[string]string{
|
|
c.Password: "[password]",
|
|
c.PemBundle: "[pem_bundle]",
|
|
c.PemJSON: "[pem_json]",
|
|
}
|
|
}
|