Addresses comments on original PR

physical/cassandra.go

@@ -1,14 +1,18 @@
 package physical
 
 import (
+	"crypto/tls"
 	"fmt"
+	"io/ioutil"
+	"strconv"
 	"strings"
 	"time"
 
 	log "github.com/mgutz/logxi/v1"
 
 	"github.com/armon/go-metrics"
 	"github.com/gocql/gocql"
+	"github.com/hashicorp/vault/helper/certutil"
 )
 
 // CassandraBackend is a physical backend that stores data in Cassandra.
@@ -72,6 +76,39 @@ func newCassandraBackend(conf map[string]string, logger log.Logger) (Backend, er
 	connectStart := time.Now()
 	cluster := gocql.NewCluster(hosts...)
 	cluster.Keyspace = keyspace
+
+	cluster.ProtoVersion = 2
+	if protoVersionStr, ok := conf["protocol_version"]; ok {
+		protoVersion, err := strconv.Atoi(protoVersionStr)
+		if err != nil {
+			return nil, fmt.Errorf("'protocol_version' must be an integer")
+		}
+		cluster.ProtoVersion = protoVersion
+	}
+
+	if username, ok := conf["username"]; ok {
+		if cluster.ProtoVersion < 2 {
+			return nil, fmt.Errorf("Authentication is not supported with protocol version < 2")
+		}
+		authenticator := gocql.PasswordAuthenticator{Username: username}
+		if password, ok := conf["password"]; ok {
+			authenticator.Password = password
+		}
+		cluster.Authenticator = authenticator
+	}
+
+	if connTimeoutStr, ok := conf["connection_timeout"]; ok {
+		connectionTimeout, err := strconv.Atoi(connTimeoutStr)
+		if err != nil {
+			return nil, fmt.Errorf("'connection_timeout' must be an integer")
+		}
+		cluster.Timeout = time.Duration(connectionTimeout) * time.Second
+	}
+
+	if err := setupCassandraTLS(conf, cluster); err != nil {
+		return nil, err
+	}
+
 	sess, err := cluster.CreateSession()
 	if err != nil {
 		return nil, err
@@ -86,6 +123,84 @@ func newCassandraBackend(conf map[string]string, logger log.Logger) (Backend, er
 	return impl, nil
 }
 
+func setupCassandraTLS(conf map[string]string, cluster *gocql.ClusterConfig) error {
+	tlsOnStr, ok := conf["tls"]
+	if !ok {
+		return nil
+	}
+
+	tlsOn, err := strconv.Atoi(tlsOnStr)
+	if err != nil {
+		return fmt.Errorf("'tls' must be an integer (0 or 1)")
+	}
+
+	if tlsOn == 0 {
+		return nil
+	}
+
+	var tlsConfig = &tls.Config{}
+	if pemBundlePath, ok := conf["pem_bundle_file"]; ok {
+		pemBundleData, err := ioutil.ReadFile(pemBundlePath)
+		if err != nil {
+			return fmt.Errorf("Error reading pem bundle from %s: %v", pemBundlePath, err)
+		}
+		pemBundle, err := certutil.ParsePEMBundle(string(pemBundleData))
+		if err != nil {
+			return fmt.Errorf("Error parsing 'pem_bundle': %v", err)
+		}
+		tlsConfig, err = pemBundle.GetTLSConfig(certutil.TLSClient)
+		if err != nil {
+			return err
+		}
+	} else {
+		if pemJSONPath, ok := conf["pem_json_file"]; ok {
+			pemJSONData, err := ioutil.ReadFile(pemJSONPath)
+			if err != nil {
+				return fmt.Errorf("Error reading json bundle from %s: %v", pemJSONPath, err)
+			}
+			pemJSON, err := certutil.ParsePKIJSON([]byte(pemJSONData))
+			if err != nil {
+				return err
+			}
+			tlsConfig, err = pemJSON.GetTLSConfig(certutil.TLSClient)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	if tlsSkipVerifyStr, ok := conf["tls_skip_verify"]; ok {
+		tlsSkipVerify, err := strconv.Atoi(tlsSkipVerifyStr)
+		if err != nil {
+			return fmt.Errorf("'tls_skip_verify' must be an integer (0 or 1)")
+		}
+		if tlsSkipVerify == 0 {
+			tlsConfig.InsecureSkipVerify = false
+		} else {
+			tlsConfig.InsecureSkipVerify = true
+		}
+	}
+
+	if tlsMinVersion, ok := conf["tls_min_version"]; ok {
+		switch tlsMinVersion {
+		case "tls10":
+			tlsConfig.MinVersion = tls.VersionTLS10
+		case "tls11":
+			tlsConfig.MinVersion = tls.VersionTLS11
+		case "tls12":
+			tlsConfig.MinVersion = tls.VersionTLS12
+		default:
+			return fmt.Errorf("'tls_min_version' must be one of `tls10`, `tls11` or `tls12`")
+		}
+	}
+
+	cluster.SslOpts = &gocql.SslOptions{
+		Config: *tlsConfig.Clone(),
+	}
+
+	return nil
+}
+
 // bucketName sanitises a bucket name for Cassandra
 func (c *CassandraBackend) bucketName(name string) string {
 	if name == "" {
@@ -164,7 +279,6 @@ func (c *CassandraBackend) List(prefix string) ([]string, error) {
 
 	stmt := fmt.Sprintf(`SELECT key FROM "%s" WHERE bucket = ?`, c.table)
 	q := c.sess.Query(stmt, c.bucketName(prefix))
-	q = q.Consistency(gocql.One) // List does not need such strong consistency guarantees
 	iter := q.Iter()
 	k, keys := "", []string{}
 	for iter.Scan(&k) {

website/source/docs/config/index.html.md

@@ -758,6 +758,35 @@ The Cassandra backend has the following options:
     in Cassandra. If set, must be one of `"ANY"`, `"ONE"`, `"TWO"`, `"THREE"`, `"QUORUM"`,
     `"ALL"`, `"LOCAL_QUORUM"`, `"EACH_QUORUM"`, or `"LOCAL_ONE"`. Defaults to `"LOCAL_QUORUM"`.
 
+  * `protocol_version` (optional) - Cassandra protocol version to use. Defaults
+    to `2`.
+
+  * `username` (optional) - Username to use when authenticating with the
+    Cassandra hosts.
+
+  * `password` (optional) - Password to use when authenticating with the
+    Cassandra hosts.
+
+  * `connection_timeout` (optional) - A timeout in seconds to wait until a
+    connection is established with the Cassandra hosts.
+
+  * `tls` (optional) - Indicates the connection with the Cassandra hosts should
+    use TLS.
+
+  * `pem_bundle_file` (optional) - Specifies a file containing a
+    certificate and private key; a certificate, private key, and issuing CA
+    certificate; or just a CA certificate.
+
+  * `pem_json_file` (optional) - Specifies a JSON file containing a certificate
+    and private key; a certificate, private key, and issuing CA certificate;
+    or just a CA certificate.
+
+  * `tls_skip_verify` (optional) - If set, then TLS host verification
+    will be disabled for Cassandra.  Defaults to `0`.
+
+  * `tls_min_version` (optional) - Minimum TLS version to use. Accepted values
+    are `tls10`, `tls11` or `tls12`. Defaults to `tls12`.
+
 You need to ensure the keyspace and table exist in Cassandra:
 
 ```cql