DBPW - Update Cassandra to adhere to v5 Database interface (#10051)

This commit is contained in:
Michael Golowka
2020-10-12 14:46:17 -06:00
committed by GitHub
parent ee6391b691
commit 1eff3f7daa
3 changed files with 270 additions and 237 deletions

View File

@@ -2,37 +2,36 @@ package cassandra
import (
"context"
"fmt"
"strings"
"time"
"github.com/gocql/gocql"
multierror "github.com/hashicorp/go-multierror"
"github.com/hashicorp/vault/api"
"github.com/hashicorp/vault/sdk/database/dbplugin"
"github.com/hashicorp/vault/sdk/database/helper/credsutil"
"github.com/hashicorp/vault/sdk/database/helper/dbutil"
"github.com/hashicorp/vault/sdk/database/newdbplugin"
"github.com/hashicorp/vault/sdk/helper/strutil"
)
const (
defaultUserCreationCQL = `CREATE USER '{{username}}' WITH PASSWORD '{{password}}' NOSUPERUSER;`
defaultUserDeletionCQL = `DROP USER '{{username}}';`
defaultRootCredentialRotationCQL = `ALTER USER {{username}} WITH PASSWORD '{{password}}';`
defaultChangePasswordCQL = `ALTER USER {{username}} WITH PASSWORD '{{password}}';`
cassandraTypeName = "cassandra"
)
var _ dbplugin.Database = &Cassandra{}
var _ newdbplugin.Database = &Cassandra{}
// Cassandra is an implementation of Database interface
type Cassandra struct {
*cassandraConnectionProducer
credsutil.CredentialsProducer
}
// New returns a new Cassandra instance
func New() (interface{}, error) {
db := new()
dbType := dbplugin.NewDatabaseErrorSanitizerMiddleware(db, db.secretValues)
dbType := newdbplugin.NewDatabaseErrorSanitizerMiddleware(db, db.secretValues)
return dbType, nil
}
@@ -41,16 +40,8 @@ func new() *Cassandra {
connProducer := &cassandraConnectionProducer{}
connProducer.Type = cassandraTypeName
credsProducer := &credsutil.SQLCredentialsProducer{
DisplayNameLen: 15,
RoleNameLen: 15,
UsernameLen: 100,
Separator: "_",
}
return &Cassandra{
cassandraConnectionProducer: connProducer,
CredentialsProducer: credsProducer,
}
}
@@ -61,7 +52,7 @@ func Run(apiTLSConfig *api.TLSConfig) error {
return err
}
dbplugin.Serve(dbType.(dbplugin.Database), api.VaultPluginTLSProvider(apiTLSConfig))
newdbplugin.Serve(dbType.(newdbplugin.Database), api.VaultPluginTLSProvider(apiTLSConfig))
return nil
}
@@ -80,45 +71,39 @@ func (c *Cassandra) getConnection(ctx context.Context) (*gocql.Session, error) {
return session.(*gocql.Session), nil
}
// CreateUser generates the username/password on the underlying Cassandra secret backend as instructed by
// the CreationStatement provided.
func (c *Cassandra) CreateUser(ctx context.Context, statements dbplugin.Statements, usernameConfig dbplugin.UsernameConfig, expiration time.Time) (username string, password string, err error) {
// Grab the lock
// NewUser generates the username/password on the underlying Cassandra secret backend as instructed by
// the statements provided.
func (c *Cassandra) NewUser(ctx context.Context, req newdbplugin.NewUserRequest) (newdbplugin.NewUserResponse, error) {
c.Lock()
defer c.Unlock()
statements = dbutil.StatementCompatibilityHelper(statements)
// Get the connection
session, err := c.getConnection(ctx)
if err != nil {
return "", "", err
return newdbplugin.NewUserResponse{}, err
}
creationCQL := statements.Creation
creationCQL := req.Statements.Commands
if len(creationCQL) == 0 {
creationCQL = []string{defaultUserCreationCQL}
}
rollbackCQL := statements.Rollback
rollbackCQL := req.RollbackStatements.Commands
if len(rollbackCQL) == 0 {
rollbackCQL = []string{defaultUserDeletionCQL}
}
username, err = c.GenerateUsername(usernameConfig)
username = strings.Replace(username, "-", "_", -1)
username, err := credsutil.GenerateUsername(
credsutil.DisplayName(req.UsernameConfig.DisplayName, 15),
credsutil.RoleName(req.UsernameConfig.RoleName, 15),
credsutil.Separator("_"),
credsutil.MaxLength(100),
credsutil.ToLower(),
)
if err != nil {
return "", "", err
return newdbplugin.NewUserResponse{}, err
}
// Cassandra doesn't like the uppercase usernames
username = strings.ToLower(username)
username = strings.ReplaceAll(username, "-", "_")
password, err = c.GeneratePassword()
if err != nil {
return "", "", err
}
// Execute each query
for _, stmt := range creationCQL {
for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
query = strings.TrimSpace(query)
@@ -126,11 +111,31 @@ func (c *Cassandra) CreateUser(ctx context.Context, statements dbplugin.Statemen
continue
}
err = session.Query(dbutil.QueryHelper(query, map[string]string{
m := map[string]string{
"username": username,
"password": password,
})).WithContext(ctx).Exec()
"password": req.Password,
}
err = session.
Query(dbutil.QueryHelper(query, m)).
WithContext(ctx).
Exec()
if err != nil {
rollbackErr := rollbackUser(ctx, session, username, rollbackCQL)
if rollbackErr != nil {
err = multierror.Append(err, rollbackErr)
}
return newdbplugin.NewUserResponse{}, err
}
}
}
resp := newdbplugin.NewUserResponse{
Username: username,
}
return resp, nil
}
func rollbackUser(ctx context.Context, session *gocql.Session, username string, rollbackCQL []string) error {
for _, stmt := range rollbackCQL {
for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
query = strings.TrimSpace(query)
@@ -138,39 +143,79 @@ func (c *Cassandra) CreateUser(ctx context.Context, statements dbplugin.Statemen
continue
}
session.Query(dbutil.QueryHelper(query, map[string]string{
m := map[string]string{
"username": username,
})).WithContext(ctx).Exec()
}
}
return "", "", err
err := session.
Query(dbutil.QueryHelper(query, m)).
WithContext(ctx).
Exec()
if err != nil {
return fmt.Errorf("failed to roll back user %s: %w", username, err)
}
}
}
return username, password, nil
}
// RenewUser is not supported on Cassandra, so this is a no-op.
func (c *Cassandra) RenewUser(ctx context.Context, statements dbplugin.Statements, username string, expiration time.Time) error {
// NOOP
return nil
}
// RevokeUser attempts to drop the specified user.
func (c *Cassandra) RevokeUser(ctx context.Context, statements dbplugin.Statements, username string) error {
// Grab the lock
c.Lock()
defer c.Unlock()
func (c *Cassandra) UpdateUser(ctx context.Context, req newdbplugin.UpdateUserRequest) (newdbplugin.UpdateUserResponse, error) {
if req.Password == nil && req.Expiration == nil {
return newdbplugin.UpdateUserResponse{}, fmt.Errorf("no changes requested")
}
statements = dbutil.StatementCompatibilityHelper(statements)
if req.Password != nil {
err := c.changeUserPassword(ctx, req.Username, req.Password)
return newdbplugin.UpdateUserResponse{}, err
}
// Expiration is no-op
return newdbplugin.UpdateUserResponse{}, nil
}
func (c *Cassandra) changeUserPassword(ctx context.Context, username string, changePass *newdbplugin.ChangePassword) error {
session, err := c.getConnection(ctx)
if err != nil {
return err
}
revocationCQL := statements.Revocation
rotateCQL := changePass.Statements.Commands
if len(rotateCQL) == 0 {
rotateCQL = []string{defaultChangePasswordCQL}
}
var result *multierror.Error
for _, stmt := range rotateCQL {
for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
query = strings.TrimSpace(query)
if len(query) == 0 {
continue
}
m := map[string]string{
"username": username,
"password": changePass.NewPassword,
}
err := session.
Query(dbutil.QueryHelper(query, m)).
WithContext(ctx).
Exec()
result = multierror.Append(result, err)
}
}
return result.ErrorOrNil()
}
// DeleteUser attempts to drop the specified user.
func (c *Cassandra) DeleteUser(ctx context.Context, req newdbplugin.DeleteUserRequest) (newdbplugin.DeleteUserResponse, error) {
c.Lock()
defer c.Unlock()
session, err := c.getConnection(ctx)
if err != nil {
return newdbplugin.DeleteUserResponse{}, err
}
revocationCQL := req.Statements.Commands
if len(revocationCQL) == 0 {
revocationCQL = []string{defaultUserDeletionCQL}
}
@@ -183,59 +228,17 @@ func (c *Cassandra) RevokeUser(ctx context.Context, statements dbplugin.Statemen
continue
}
err := session.Query(dbutil.QueryHelper(query, map[string]string{
"username": username,
})).WithContext(ctx).Exec()
m := map[string]string{
"username": req.Username,
}
err := session.
Query(dbutil.QueryHelper(query, m)).
WithContext(ctx).
Exec()
result = multierror.Append(result, err)
}
}
return result.ErrorOrNil()
}
func (c *Cassandra) RotateRootCredentials(ctx context.Context, statements []string) (map[string]interface{}, error) {
// Grab the lock
c.Lock()
defer c.Unlock()
session, err := c.getConnection(ctx)
if err != nil {
return nil, err
}
rotateCQL := statements
if len(rotateCQL) == 0 {
rotateCQL = []string{defaultRootCredentialRotationCQL}
}
password, err := c.GeneratePassword()
if err != nil {
return nil, err
}
var result *multierror.Error
for _, stmt := range rotateCQL {
for _, query := range strutil.ParseArbitraryStringSlice(stmt, ";") {
query = strings.TrimSpace(query)
if len(query) == 0 {
continue
}
err := session.Query(dbutil.QueryHelper(query, map[string]string{
"username": c.Username,
"password": password,
})).WithContext(ctx).Exec()
result = multierror.Append(result, err)
}
}
err = result.ErrorOrNil()
if err != nil {
return nil, err
}
c.rawConfig["password"] = password
return c.rawConfig, nil
return newdbplugin.DeleteUserResponse{}, result.ErrorOrNil()
}

View File

@@ -1,22 +1,38 @@
package cassandra
import (
"context"
"reflect"
"regexp"
"strings"
"testing"
"time"
dbtesting "github.com/hashicorp/vault/sdk/database/newdbplugin/testing"
backoff "github.com/cenkalti/backoff/v3"
"github.com/gocql/gocql"
"github.com/hashicorp/errwrap"
"github.com/hashicorp/vault/helper/testhelpers/cassandra"
"github.com/hashicorp/vault/sdk/database/dbplugin"
"github.com/hashicorp/vault/sdk/database/newdbplugin"
)
func getCassandra(t *testing.T, protocolVersion interface{}) (*Cassandra, func()) {
cleanup, connURL := cassandra.PrepareTestContainer(t, "latest")
pieces := strings.Split(connURL, ":")
connectionDetails := map[string]interface{}{
db := new()
initReq := newdbplugin.InitializeRequest{
Config: map[string]interface{}{
"hosts": connURL,
"port": pieces[1],
"username": "cassandra",
"password": "cassandra",
"protocol_version": protocolVersion,
"connect_timeout": "20s",
},
VerifyConnection: true,
}
expectedConfig := map[string]interface{}{
"hosts": connURL,
"port": pieces[1],
"username": "cassandra",
@@ -25,10 +41,9 @@ func getCassandra(t *testing.T, protocolVersion interface{}) (*Cassandra, func()
"connect_timeout": "20s",
}
db := new()
_, err := db.Init(context.Background(), connectionDetails, true)
if err != nil {
t.Fatalf("err: %s", err)
initResp := dbtesting.AssertInitialize(t, db, initReq)
if !reflect.DeepEqual(initResp.Config, expectedConfig) {
t.Fatalf("Initialize response config actual: %#v\nExpected: %#v", initResp.Config, expectedConfig)
}
if !db.Initialized {
@@ -54,109 +69,115 @@ func TestCassandra_CreateUser(t *testing.T) {
db, cleanup := getCassandra(t, 4)
defer cleanup()
statements := dbplugin.Statements{
Creation: []string{testCassandraRole},
}
usernameConfig := dbplugin.UsernameConfig{
password := "myreallysecurepassword"
createReq := newdbplugin.NewUserRequest{
UsernameConfig: newdbplugin.UsernameMetadata{
DisplayName: "test",
RoleName: "test",
},
Statements: newdbplugin.Statements{
Commands: []string{createUserStatements},
},
Password: password,
Expiration: time.Now().Add(1 * time.Minute),
}
username, password, err := db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(time.Minute))
if err != nil {
t.Fatalf("err: %s", err)
createResp := dbtesting.AssertNewUser(t, db, createReq)
expectedRegex := "^v_test_test_[a-zA-Z0-9]{20}_[0-9]{10}$"
re := regexp.MustCompile(expectedRegex)
if !re.MatchString(createResp.Username) {
t.Fatalf("Generated username %q did not match regexp %q", createResp.Username, expectedRegex)
}
if err := testCredsExist(db.Hosts, db.Port, username, password); err != nil {
t.Fatalf("Could not connect with new credentials: %s", err)
}
assertCreds(t, db.Hosts, db.Port, createResp.Username, password, 5*time.Second)
}
func TestMyCassandra_RenewUser(t *testing.T) {
func TestMyCassandra_UpdateUserPassword(t *testing.T) {
db, cleanup := getCassandra(t, 4)
defer cleanup()
statements := dbplugin.Statements{
Creation: []string{testCassandraRole},
}
usernameConfig := dbplugin.UsernameConfig{
password := "myreallysecurepassword"
createReq := newdbplugin.NewUserRequest{
UsernameConfig: newdbplugin.UsernameMetadata{
DisplayName: "test",
RoleName: "test",
},
Statements: newdbplugin.Statements{
Commands: []string{createUserStatements},
},
Password: password,
Expiration: time.Now().Add(1 * time.Minute),
}
username, password, err := db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(time.Minute))
if err != nil {
t.Fatalf("err: %s", err)
createResp := dbtesting.AssertNewUser(t, db, createReq)
assertCreds(t, db.Hosts, db.Port, createResp.Username, password, 5*time.Second)
newPassword := "somenewpassword"
updateReq := newdbplugin.UpdateUserRequest{
Username: createResp.Username,
Password: &newdbplugin.ChangePassword{
NewPassword: newPassword,
Statements: newdbplugin.Statements{},
},
Expiration: nil,
}
if err := testCredsExist(db.Hosts, db.Port, username, password); err != nil {
t.Fatalf("Could not connect with new credentials: %s", err)
dbtesting.AssertUpdateUser(t, db, updateReq)
assertCreds(t, db.Hosts, db.Port, createResp.Username, newPassword, 5*time.Second)
}
err = db.RenewUser(context.Background(), statements, username, time.Now().Add(time.Minute))
if err != nil {
t.Fatalf("err: %s", err)
}
}
func TestCassandra_RevokeUser(t *testing.T) {
func TestCassandra_DeleteUser(t *testing.T) {
db, cleanup := getCassandra(t, 4)
defer cleanup()
statements := dbplugin.Statements{
Creation: []string{testCassandraRole},
}
usernameConfig := dbplugin.UsernameConfig{
password := "myreallysecurepassword"
createReq := newdbplugin.NewUserRequest{
UsernameConfig: newdbplugin.UsernameMetadata{
DisplayName: "test",
RoleName: "test",
},
Statements: newdbplugin.Statements{
Commands: []string{createUserStatements},
},
Password: password,
Expiration: time.Now().Add(1 * time.Minute),
}
username, password, err := db.CreateUser(context.Background(), statements, usernameConfig, time.Now().Add(time.Minute))
createResp := dbtesting.AssertNewUser(t, db, createReq)
assertCreds(t, db.Hosts, db.Port, createResp.Username, password, 5*time.Second)
deleteReq := newdbplugin.DeleteUserRequest{
Username: createResp.Username,
}
dbtesting.AssertDeleteUser(t, db, deleteReq)
assertNoCreds(t, db.Hosts, db.Port, createResp.Username, password, 5*time.Second)
}
func assertCreds(t testing.TB, address string, port int, username, password string, timeout time.Duration) {
t.Helper()
op := func() error {
return connect(t, address, port, username, password)
}
bo := backoff.NewExponentialBackOff()
bo.MaxElapsedTime = timeout
bo.InitialInterval = 500 * time.Millisecond
bo.MaxInterval = bo.InitialInterval
bo.RandomizationFactor = 0.0
err := backoff.Retry(op, bo)
if err != nil {
t.Fatalf("err: %s", err)
}
if err = testCredsExist(db.Hosts, db.Port, username, password); err != nil {
t.Fatalf("Could not connect with new credentials: %s", err)
}
// Test default revoke statements
err = db.RevokeUser(context.Background(), statements, username)
if err != nil {
t.Fatalf("err: %s", err)
}
if err = testCredsExist(db.Hosts, db.Port, username, password); err == nil {
t.Fatal("Credentials were not revoked")
t.Fatalf("failed to connect after %s: %s", timeout, err)
}
}
func TestCassandra_RotateRootCredentials(t *testing.T) {
db, cleanup := getCassandra(t, 4)
defer cleanup()
if !db.cassandraConnectionProducer.Initialized {
t.Fatal("Database should be initialized")
}
newConf, err := db.RotateRootCredentials(context.Background(), nil)
if err != nil {
t.Fatalf("err: %v", err)
}
if newConf["password"] == "cassandra" {
t.Fatal("password was not updated")
}
err = db.Close()
if err != nil {
t.Fatalf("err: %s", err)
}
}
func testCredsExist(address string, port int, username, password string) error {
func connect(t testing.TB, address string, port int, username, password string) error {
t.Helper()
clusterConfig := gocql.NewCluster(address)
clusterConfig.Authenticator = gocql.PasswordAuthenticator{
Username: username,
@@ -167,11 +188,34 @@ func testCredsExist(address string, port int, username, password string) error {
session, err := clusterConfig.CreateSession()
if err != nil {
return errwrap.Wrapf("error creating session: {{err}}", err)
return err
}
defer session.Close()
return nil
}
const testCassandraRole = `CREATE USER '{{username}}' WITH PASSWORD '{{password}}' NOSUPERUSER;
func assertNoCreds(t testing.TB, address string, port int, username, password string, timeout time.Duration) {
t.Helper()
op := func() error {
// "Invert" the error so the backoff logic sees a failure to connect as a success
err := connect(t, address, port, username, password)
if err != nil {
return nil
}
return err
}
bo := backoff.NewExponentialBackOff()
bo.MaxElapsedTime = timeout
bo.InitialInterval = 500 * time.Millisecond
bo.MaxInterval = bo.InitialInterval
bo.RandomizationFactor = 0.0
err := backoff.Retry(op, bo)
if err != nil {
t.Fatalf("successfully connected after %s when it shouldn't", timeout)
}
}
const createUserStatements = `CREATE USER '{{username}}' WITH PASSWORD '{{password}}' NOSUPERUSER;
GRANT ALL PERMISSIONS ON ALL KEYSPACES TO {{username}};`

View File

@@ -12,9 +12,9 @@ import (
"github.com/gocql/gocql"
"github.com/hashicorp/errwrap"
"github.com/hashicorp/vault/sdk/database/dbplugin"
"github.com/hashicorp/vault/sdk/database/helper/connutil"
"github.com/hashicorp/vault/sdk/database/helper/dbutil"
"github.com/hashicorp/vault/sdk/database/newdbplugin"
"github.com/hashicorp/vault/sdk/helper/certutil"
"github.com/hashicorp/vault/sdk/helper/parseutil"
"github.com/hashicorp/vault/sdk/helper/tlsutil"
@@ -52,20 +52,15 @@ type cassandraConnectionProducer struct {
sync.Mutex
}
func (c *cassandraConnectionProducer) Initialize(ctx context.Context, conf map[string]interface{}, verifyConnection bool) error {
_, err := c.Init(ctx, conf, verifyConnection)
return err
}
func (c *cassandraConnectionProducer) Init(ctx context.Context, conf map[string]interface{}, verifyConnection bool) (map[string]interface{}, error) {
func (c *cassandraConnectionProducer) Initialize(ctx context.Context, req newdbplugin.InitializeRequest) (newdbplugin.InitializeResponse, error) {
c.Lock()
defer c.Unlock()
c.rawConfig = conf
c.rawConfig = req.Config
err := mapstructure.WeakDecode(conf, c)
err := mapstructure.WeakDecode(req.Config, c)
if err != nil {
return nil, err
return newdbplugin.InitializeResponse{}, err
}
if c.ConnectTimeoutRaw == nil {
@@ -73,7 +68,7 @@ func (c *cassandraConnectionProducer) Init(ctx context.Context, conf map[string]
}
c.connectTimeout, err = parseutil.ParseDurationSecond(c.ConnectTimeoutRaw)
if err != nil {
return nil, errwrap.Wrapf("invalid connect_timeout: {{err}}", err)
return newdbplugin.InitializeResponse{}, errwrap.Wrapf("invalid connect_timeout: {{err}}", err)
}
if c.SocketKeepAliveRaw == nil {
@@ -81,16 +76,16 @@ func (c *cassandraConnectionProducer) Init(ctx context.Context, conf map[string]
}
c.socketKeepAlive, err = parseutil.ParseDurationSecond(c.SocketKeepAliveRaw)
if err != nil {
return nil, errwrap.Wrapf("invalid socket_keep_alive: {{err}}", err)
return newdbplugin.InitializeResponse{}, errwrap.Wrapf("invalid socket_keep_alive: {{err}}", err)
}
switch {
case len(c.Hosts) == 0:
return nil, fmt.Errorf("hosts cannot be empty")
return newdbplugin.InitializeResponse{}, fmt.Errorf("hosts cannot be empty")
case len(c.Username) == 0:
return nil, fmt.Errorf("username cannot be empty")
return newdbplugin.InitializeResponse{}, fmt.Errorf("username cannot be empty")
case len(c.Password) == 0:
return nil, fmt.Errorf("password cannot be empty")
return newdbplugin.InitializeResponse{}, fmt.Errorf("password cannot be empty")
}
var certBundle *certutil.CertBundle
@@ -99,11 +94,11 @@ func (c *cassandraConnectionProducer) Init(ctx context.Context, conf map[string]
case len(c.PemJSON) != 0:
parsedCertBundle, err = certutil.ParsePKIJSON([]byte(c.PemJSON))
if err != nil {
return nil, errwrap.Wrapf("could not parse given JSON; it must be in the format of the output of the PKI backend certificate issuing command: {{err}}", err)
return newdbplugin.InitializeResponse{}, errwrap.Wrapf("could not parse given JSON; it must be in the format of the output of the PKI backend certificate issuing command: {{err}}", err)
}
certBundle, err = parsedCertBundle.ToCertBundle()
if err != nil {
return nil, errwrap.Wrapf("Error marshaling PEM information: {{err}}", err)
return newdbplugin.InitializeResponse{}, errwrap.Wrapf("Error marshaling PEM information: {{err}}", err)
}
c.certificate = certBundle.Certificate
c.privateKey = certBundle.PrivateKey
@@ -113,11 +108,11 @@ func (c *cassandraConnectionProducer) Init(ctx context.Context, conf map[string]
case len(c.PemBundle) != 0:
parsedCertBundle, err = certutil.ParsePEMBundle(c.PemBundle)
if err != nil {
return nil, errwrap.Wrapf("Error parsing the given PEM information: {{err}}", err)
return newdbplugin.InitializeResponse{}, errwrap.Wrapf("Error parsing the given PEM information: {{err}}", err)
}
certBundle, err = parsedCertBundle.ToCertBundle()
if err != nil {
return nil, errwrap.Wrapf("Error marshaling PEM information: {{err}}", err)
return newdbplugin.InitializeResponse{}, errwrap.Wrapf("Error marshaling PEM information: {{err}}", err)
}
c.certificate = certBundle.Certificate
c.privateKey = certBundle.PrivateKey
@@ -129,13 +124,17 @@ func (c *cassandraConnectionProducer) Init(ctx context.Context, conf map[string]
// and the connection can be established at a later time.
c.Initialized = true
if verifyConnection {
if req.VerifyConnection {
if _, err := c.Connection(ctx); err != nil {
return nil, errwrap.Wrapf("error verifying connection: {{err}}", err)
return newdbplugin.InitializeResponse{}, errwrap.Wrapf("error verifying connection: {{err}}", err)
}
}
return conf, nil
resp := newdbplugin.InitializeResponse{
Config: req.Config,
}
return resp, nil
}
func (c *cassandraConnectionProducer) Connection(ctx context.Context) (interface{}, error) {
@@ -160,7 +159,6 @@ func (c *cassandraConnectionProducer) Connection(ctx context.Context) (interface
}
func (c *cassandraConnectionProducer) Close() error {
// Grab the write lock
c.Lock()
defer c.Unlock()
@@ -246,7 +244,6 @@ func (c *cassandraConnectionProducer) createSession(ctx context.Context) (*gocql
return nil, errwrap.Wrapf("error creating session: {{err}}", err)
}
// Set consistency
if c.Consistency != "" {
consistencyValue, err := gocql.ParseConsistencyWrapper(c.Consistency)
if err != nil {
@@ -257,7 +254,6 @@ func (c *cassandraConnectionProducer) createSession(ctx context.Context) (*gocql
session.SetConsistency(consistencyValue)
}
// Verify the info
if !c.SkipVerification {
err = session.Query(`LIST ALL`).WithContext(ctx).Exec()
if err != nil && len(c.Username) != 0 && strings.Contains(err.Error(), "not authorized") {
@@ -278,20 +274,10 @@ func (c *cassandraConnectionProducer) createSession(ctx context.Context) (*gocql
return session, nil
}
func (c *cassandraConnectionProducer) secretValues() map[string]interface{} {
return map[string]interface{}{
func (c *cassandraConnectionProducer) secretValues() map[string]string {
return map[string]string{
c.Password: "[password]",
c.PemBundle: "[pem_bundle]",
c.PemJSON: "[pem_json]",
}
}
// SetCredentials uses provided information to set/create a user in the
// database. Unlike CreateUser, this method requires a username be provided and
// uses the name given, instead of generating a name. This is used for creating
// and setting the password of static accounts, as well as rolling back
// passwords in the database in the event an updated database fails to save in
// Vault's storage.
func (c *cassandraConnectionProducer) SetCredentials(ctx context.Context, statements dbplugin.Statements, staticUser dbplugin.StaticUserConfig) (username, password string, err error) {
return "", "", dbutil.Unimplemented()
}