github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-21 08:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(cassandra): support HOCON Kerberos and Astra natively

Browse Source
This commit is contained in:
t8y2
2026-08-04 02:59:41 +08:00
parent 6193055453
commit 6f0b4666f8
10 changed files with 2065 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/drivers/cassandra-go/README.md
+84 -6
View File
@@ -8,8 +8,14 @@ multi-session JSON-RPC protocol without a JVM.
- Native protocol versions: v3-v5
- Declared server range: Apache Cassandra 2.1+
- Live validation: 2.2.19, 3.11.19, 4.1.10, and 5.0.6
- Authentication: username/password
- Kerberos live validation: Cassandra 4.1.10 with password, keytab, FILE ccache,
JAAS discovery, and HOCON `configfile`
- Astra validation: secure-connect bundle parsing and transport configuration;
live Astra credentials were not available
- Authentication: username/password and Kerberos/GSSAPI
- TLS: CA verification, optional client certificate/key, hostname verification
- Cloud: DataStax Astra secure connect bundles
- Configuration: Java Driver 4 HOCON `configfile` mapping plus native extensions
- Metadata: keyspaces, tables, columns, indexes, CQL table DDL, completion search
- Queries: legacy string result values, paging, cancellation, logged and unlogged batches
@@ -33,15 +39,87 @@ syntax.
| `sslenginefactory` | the standard `DefaultSslEngineFactory` maps to native TLS |
| `hostnameverification` | TLS hostname verification; enabled by default |
| `user`, `password` | password authentication |
| `configfile` | Java Driver 4 HOCON configuration; overrides URL options except contact points and keyspace |
| `usekrb5` | Kerberos/GSSAPI authentication using password, keytab, or FILE credential cache |
| `secureconnectbundle` | DataStax Astra secure connect bundle; contact points and manual TLS options are ignored |
| `requesttimeout`, `connecttimeout` | request and connection deadlines |
| `tcpnodelay`, `keepalive` | native TCP socket options |
| `compliancemode` | accepted; JDBC-only `java.sql` behavior is not applicable to JSON-RPC |
Java implementation hooks do not have a safe native equivalent. The Agent
returns a targeted connection error for `configfile`, `usekrb5=true`,
`secureconnectbundle`, custom `sslenginefactory` classes, and custom policy
classes. Translate Java HOCON settings to the supported URL parameters before
migrating a connection.
The Agent rejects custom Java implementation classes because they cannot be
loaded by a native binary. This includes custom authentication, SSL, retry,
reconnection, and load-balancing classes. Java JKS/PKCS12 truststores and
keystores are not read directly; use the native PEM paths described below.
## Java Driver HOCON configuration
`configfile` reads Java Driver 4 HOCON files and preserves the JDBC wrapper's
precedence: the file overrides URL options except contact points and keyspace.
A missing file is ignored for compatibility with the JDBC wrapper.
Mapped Java Driver paths include:
- `basic.request.timeout`, `consistency`, `serial-consistency`, and `page-size`
- `basic.load-balancing-policy.class` and `local-datacenter`
- `basic.cloud.secure-connect-bundle`
- `advanced.connection.connect-timeout` and `pool.local.size`
- `advanced.socket.tcp-no-delay` and `keep-alive`
- `advanced.protocol.version`, retry policy, and reconnection policy
- `advanced.auth-provider` plaintext and Instaclustr Kerberos options
- `advanced.ssl-engine-factory` default TLS and hostname validation
Native-only settings can be placed under `dbx.cassandra`:
```hocon
dbx.cassandra {
tls {
enabled = true
ca-cert-path = "/path/to/ca.pem"
client-cert-path = "/path/to/client.pem"
client-key-path = "/path/to/client-key.pem"
hostname-verification = true
}
kerberos {
enabled = true
config = "/etc/krb5.conf"
jaas-config = "/path/to/jaas.conf"
principal = "alice@EXAMPLE.COM"
keytab = "/path/to/alice.keytab"
service-name = "cassandra"
server-name = "node1.example.com"
authorization-id = "assumed_role"
qop = "auth"
}
}
```
## Kerberos
`usekrb5=true` implements the same GSSAPI flow used by the former Instaclustr
Java auth provider. The service principal defaults to
`cassandra/<canonical-node-hostname>`. Set `kerberosservername` when reverse DNS
does not resolve to the service-principal hostname.
Credential discovery order is:
1. Explicit JAAS `CassandraJavaClient` cache/keytab selection
2. Explicit `kerberosccache` or `kerberoskeytab`
3. Explicit principal and password
4. `KRB5CCNAME`, then `KRB5_CLIENT_KTNAME`/`KRB5_KTNAME`
The Agent also reads `java.security.auth.login.config` and
`java.security.krb5.conf` from `JAVA_TOOL_OPTIONS`, `_JAVA_OPTIONS`, or
`JDK_JAVA_OPTIONS`. Only FILE credential caches are supported. SASL QOP `auth`
is supported; `auth-int` and `auth-conf` are rejected because they require
wrapping Cassandra traffic after authentication.
## Astra secure connect bundles
Set `secureconnectbundle` to a local Astra bundle ZIP and provide its database
credentials with `user` and `password`. A normal Cassandra host is not required.
Kerberos cannot be combined with a secure connect bundle. Manual TLS settings
are ignored because the bundle supplies its own CA, client certificate, key,
SNI endpoint, and metadata service.
## Integration test
agents/drivers/cassandra-go/astra_test.go
+166
View File
@@ -0,0 +1,166 @@
package main
import (
"archive/zip"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/x509"
"crypto/x509/pkix"
"encoding/json"
"encoding/pem"
"math/big"
"net/url"
"os"
"path/filepath"
"reflect"
"testing"
"time"
gocql "github.com/apache/cassandra-gocql-driver/v2"
)
func TestSecureConnectBundleBuildsAstraClusterWithoutHost(t *testing.T) {
bundlePath := writeTestSecureConnectBundle(t)
config, err := parseCassandraConfig(connectParams{
Username: "token",
Password: "astra-token",
URLParams: url.Values{
"secureconnectbundle": []string{bundlePath},
"requesttimeout": []string{"9s"},
"connecttimeout": []string{"7s"},
}.Encode(),
})
if err != nil {
t.Fatal(err)
}
if len(config.hosts) != 0 || config.secureConnectBundle != bundlePath {
t.Fatalf("unexpected Astra config: %#v", config)
}
cluster, err := config.clusterConfig("app")
if err != nil {
t.Fatal(err)
}
if cluster.HostDialer == nil {
t.Fatal("Astra cluster must use the secure-connect HostDialer")
}
if reflect.DeepEqual(cluster.Hosts, config.hosts) || len(cluster.Hosts) != 3 {
t.Fatalf("Astra cluster must use dialer placeholder hosts: %#v", cluster.Hosts)
}
credentials, ok := cluster.Authenticator.(*gocql.PasswordAuthenticator)
if !ok || credentials.Username != "token" || credentials.Password != "astra-token" {
t.Fatalf("unexpected Astra authenticator: %#v", cluster.Authenticator)
}
if cluster.Keyspace != "app" || cluster.Timeout != 9*time.Second || cluster.ConnectTimeout != 7*time.Second {
t.Fatalf("unexpected Astra cluster options: %#v", cluster)
}
}
func TestSecureConnectBundleSupportsHOCONConfiguration(t *testing.T) {
bundlePath := writeTestSecureConnectBundle(t)
configPath := writeTestFile(t, "astra.conf", `
datastax-java-driver {
basic.cloud.secure-connect-bundle = "`+bundlePath+`"
advanced.auth-provider {
class = PlainTextAuthProvider
username = token
password = astra-token
}
}
`)
config, err := parseCassandraConfig(connectParams{
URLParams: url.Values{"configfile": []string{configPath}}.Encode(),
})
if err != nil {
t.Fatal(err)
}
if config.secureConnectBundle != bundlePath || config.username != "token" || config.password != "astra-token" {
t.Fatalf("unexpected HOCON Astra config: %#v", config)
}
if _, err := config.clusterConfig(""); err != nil {
t.Fatal(err)
}
}
func TestSecureConnectBundleValidatesCredentialsAndAuthMode(t *testing.T) {
bundlePath := writeTestSecureConnectBundle(t)
if _, err := parseCassandraConfig(connectParams{
URLParams: url.Values{"secureconnectbundle": []string{bundlePath}}.Encode(),
}); err == nil {
t.Fatal("expected Astra credential validation error")
}
if _, err := parseCassandraConfig(connectParams{
Username: "token",
Password: "astra-token",
URLParams: url.Values{
"secureconnectbundle": []string{bundlePath},
"usekrb5": []string{"true"},
}.Encode(),
}); err == nil {
t.Fatal("expected Astra and Kerberos conflict")
}
}
func writeTestSecureConnectBundle(t *testing.T) string {
t.Helper()
privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
if err != nil {
t.Fatal(err)
}
template := &x509.Certificate{
SerialNumber: big.NewInt(1),
Subject: pkix.Name{CommonName: "dbx-astra-test"},
NotBefore: time.Now().Add(-time.Hour),
NotAfter: time.Now().Add(time.Hour),
IsCA: true,
BasicConstraintsValid: true,
KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth, x509.ExtKeyUsageServerAuth},
}
certificateDER, err := x509.CreateCertificate(rand.Reader, template, template, &privateKey.PublicKey, privateKey)
if err != nil {
t.Fatal(err)
}
privateKeyDER, err := x509.MarshalECPrivateKey(privateKey)
if err != nil {
t.Fatal(err)
}
certificatePEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificateDER})
privateKeyPEM := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: privateKeyDER})
configJSON, err := json.Marshal(map[string]any{"host": "astra.example.com", "port": 29042})
if err != nil {
t.Fatal(err)
}
bundlePath := filepath.Join(t.TempDir(), "secure-connect.zip")
bundle, err := os.Create(bundlePath)
if err != nil {
t.Fatal(err)
}
archive := zip.NewWriter(bundle)
for name, contents := range map[string][]byte{
"config.json": configJSON,
"ca.crt": certificatePEM,
"cert": certificatePEM,
"key": privateKeyPEM,
} {
entry, createErr := archive.Create(name)
if createErr != nil {
t.Fatal(createErr)
}
if _, writeErr := entry.Write(contents); writeErr != nil {
t.Fatal(writeErr)
}
}
if err := archive.Close(); err != nil {
t.Fatal(err)
}
if err := bundle.Close(); err != nil {
t.Fatal(err)
}
return bundlePath
}
agents/drivers/cassandra-go/config.go
+118 -25
View File
@@ -9,6 +9,7 @@ import (
"time"
gocql "github.com/apache/cassandra-gocql-driver/v2"
gocqlastra "github.com/datastax/gocql-astra/v2"
)
type cassandraConfig struct {
@@ -41,6 +42,9 @@ type cassandraConfig struct {
reconnectionMaxDelay time.Duration
loadBalancingPolicy string
disableInitialHostLookup bool
configFile string
secureConnectBundle string
kerberos kerberosConfig
}
func parseCassandraConfig(cp connectParams) (cassandraConfig, error) {
@@ -62,6 +66,7 @@ func parseCassandraConfig(cp connectParams) (cassandraConfig, error) {
retryCount: 3,
reconnectionBaseDelay: time.Second,
reconnectionMaxDelay: 60 * time.Second,
kerberos: defaultKerberosConfig(),
}
if cp.Port > 0 {
config.port = cp.Port
@@ -76,10 +81,6 @@ func parseCassandraConfig(cp connectParams) (cassandraConfig, error) {
if len(config.hosts) == 0 {
config.hosts = splitHosts(cp.Host)
}
if len(config.hosts) == 0 {
return cassandraConfig{}, fmt.Errorf("Cassandra host is required")
}
urlParams, err := parseURLParams(cp.URLParams)
if err != nil {
return cassandraConfig{}, err
@@ -90,7 +91,18 @@ func parseCassandraConfig(cp connectParams) (cassandraConfig, error) {
if err := applyCassandraURLParams(&config, params); err != nil {
return cassandraConfig{}, err
}
if !config.disableInitialHostLookup && allLoopbackHosts(config.hosts) {
if config.configFile != "" {
if err := applyCassandraConfigFile(&config, config.configFile); err != nil {
return cassandraConfig{}, err
}
}
if err := config.finalize(); err != nil {
return cassandraConfig{}, err
}
if len(config.hosts) == 0 && config.secureConnectBundle == "" {
return cassandraConfig{}, fmt.Errorf("Cassandra host is required")
}
if len(config.hosts) > 0 && !config.disableInitialHostLookup && allLoopbackHosts(config.hosts) {
config.disableInitialHostLookup = true
}
return config, nil
@@ -276,17 +288,53 @@ func applyCassandraURLParams(config *cassandraConfig, params url.Values) error {
if err != nil {
return fmt.Errorf("invalid usekrb5 option: %w", err)
}
if enabled {
return fmt.Errorf("Cassandra Kerberos authentication is not supported by the native agent")
}
config.kerberos.enabled = enabled
case "secureconnectbundle":
if value != "" {
return fmt.Errorf("Cassandra secure connect bundles are not supported by the native agent")
}
config.secureConnectBundle = value
case "configfile":
if value != "" {
return fmt.Errorf("Cassandra Java driver configfile is not supported; translate it to native URL parameters")
config.configFile = value
case "kerberosconfig", "kerberosconfigpath", "krb5config", "krb5conf":
config.kerberos.configPath = value
case "jaasconfig", "jaasconfigpath":
config.kerberos.jaasConfigPath = value
case "kerberosprincipal", "krb5principal":
config.kerberos.principal = value
case "kerberosrealm", "krb5realm":
config.kerberos.realm = value
case "kerberoskeytab", "keytab":
config.kerberos.keytabPath = value
case "kerberosccache", "kerberosticketcache", "ccache", "ticketcache":
config.kerberos.ccachePath = value
case "kerberospassword":
config.kerberos.password = value
case "kerberosservice", "kerberosservicename", "saslprotocol":
config.kerberos.serviceName = value
case "kerberosservername", "saslservername":
config.kerberos.serverName = value
case "kerberosauthorizationid", "authorizationid":
config.kerberos.authorizationID = value
case "kerberosqop", "saslqop":
config.kerberos.qop = value
case "kerberosdisablepafxfast", "disablepafxfast":
disabled, err := strconv.ParseBool(value)
if err != nil {
return fmt.Errorf("invalid disablepafxfast option: %w", err)
}
config.kerberos.disablePAFXFAST = disabled
case "kerberosusekeytab", "usekeytab":
enabled, err := strconv.ParseBool(value)
if err != nil {
return fmt.Errorf("invalid usekeytab option: %w", err)
}
config.kerberos.useKeytab = enabled
config.kerberos.useKeytabSet = true
case "kerberosuseticketcache", "useticketcache":
enabled, err := strconv.ParseBool(value)
if err != nil {
return fmt.Errorf("invalid useticketcache option: %w", err)
}
config.kerberos.useTicketCache = enabled
config.kerberos.useTicketCacheSet = true
case "compliancemode":
// JDBC compliance modes only alter java.sql behavior. The native DBX
// JSON-RPC contract already defines statement and transaction behavior.
@@ -298,21 +346,35 @@ func applyCassandraURLParams(config *cassandraConfig, params url.Values) error {
}
func (config cassandraConfig) clusterConfig(keyspace string) (*gocql.ClusterConfig, error) {
cluster := gocql.NewCluster(config.hosts...)
cluster.Port = config.port
var cluster *gocql.ClusterConfig
var err error
if config.secureConnectBundle != "" {
cluster, err = gocqlastra.NewClusterFromBundle(
config.secureConnectBundle,
config.username,
config.password,
config.connectTimeout,
)
if err != nil {
return nil, fmt.Errorf("load Cassandra secure connect bundle: %w", err)
}
} else {
cluster = gocql.NewCluster(config.hosts...)
cluster.Port = config.port
cluster.Dialer = cassandraDialer{
timeout: config.connectTimeout,
keepAlive: config.keepAlive,
tcpNoDelay: config.tcpNoDelay,
}
cluster.DisableInitialHostLookup = config.disableInitialHostLookup
cluster.IgnorePeerAddr = config.disableInitialHostLookup
}
cluster.Keyspace = strings.TrimSpace(keyspace)
cluster.Timeout = config.requestTimeout
cluster.ConnectTimeout = config.connectTimeout
cluster.WriteTimeout = config.requestTimeout
cluster.NumConns = config.numConnections
cluster.PageSize = config.pageSize
cluster.Dialer = cassandraDialer{
timeout: config.connectTimeout,
keepAlive: config.keepAlive,
tcpNoDelay: config.tcpNoDelay,
}
cluster.DisableInitialHostLookup = config.disableInitialHostLookup
cluster.IgnorePeerAddr = config.disableInitialHostLookup
if config.protocolVersion != 0 {
cluster.ProtoVersion = config.protocolVersion
}
@@ -333,10 +395,17 @@ func (config cassandraConfig) clusterConfig(keyspace string) (*gocql.ClusterConf
}
cluster.SerialConsistency = consistency
}
if config.username != "" {
if config.kerberos.enabled {
authProvider, err := newKerberosAuthProvider(config.kerberos, config.username, config.password)
if err != nil {
return nil, err
}
cluster.Authenticator = nil
cluster.AuthProvider = authProvider
} else if config.secureConnectBundle == "" && config.username != "" {
cluster.Authenticator = gocql.PasswordAuthenticator{Username: config.username, Password: config.password}
}
if config.ssl {
if config.secureConnectBundle == "" && config.ssl {
cluster.SslOpts = &gocql.SslOptions{
CaPath: config.caCertPath,
CertPath: config.clientCertPath,
@@ -356,6 +425,30 @@ func (config cassandraConfig) clusterConfig(keyspace string) (*gocql.ClusterConf
return cluster, nil
}
func (config *cassandraConfig) finalize() error {
var err error
config.configFile, err = normalizeLocalFilePath(config.configFile)
if err != nil {
return fmt.Errorf("invalid Cassandra configfile: %w", err)
}
config.secureConnectBundle, err = normalizeLocalFilePath(config.secureConnectBundle)
if err != nil {
return fmt.Errorf("invalid Cassandra secureconnectbundle: %w", err)
}
if config.secureConnectBundle != "" && config.kerberos.enabled {
return fmt.Errorf("Cassandra secure connect bundles cannot be combined with Kerberos authentication")
}
if config.secureConnectBundle != "" && (config.username == "" || config.password == "") {
return fmt.Errorf("Cassandra secure connect bundles require username and password credentials")
}
if config.kerberos.enabled {
if err := config.kerberos.finalize(config.username, config.password); err != nil {
return err
}
}
return nil
}
func splitHosts(raw string) []string {
raw = strings.ReplaceAll(raw, "--", ",")
parts := strings.FieldsFunc(raw, func(char rune) bool { return char == ',' || char == ';' })
@@ -422,7 +515,7 @@ func normalizeRetryPolicy(value string) (string, error) {
func normalizeLoadBalancingPolicy(value string) (string, error) {
name := strings.ToLower(simpleClassName(value))
switch name {
case "", "dcinferringloadbalancingpolicy", "defaultloadbalancingpolicy":
case "", "basicloadbalancingpolicy", "dcinferringloadbalancingpolicy", "defaultloadbalancingpolicy":
return "default", nil
case "roundrobinpolicy":
return "round_robin", nil
agents/drivers/cassandra-go/config_file.go
+469
View File
@@ -0,0 +1,469 @@
package main
import (
"errors"
"fmt"
"net/url"
"os"
"path/filepath"
"runtime"
"strconv"
"strings"
"time"
gocql "github.com/apache/cassandra-gocql-driver/v2"
"github.com/gurkankaymak/hocon"
)
const javaDriverConfigPrefix = "datastax-java-driver."
func applyCassandraConfigFile(config *cassandraConfig, rawPath string) error {
path, err := normalizeLocalFilePath(rawPath)
if err != nil {
return fmt.Errorf("invalid Cassandra configfile: %w", err)
}
if path == "" {
return nil
}
info, err := os.Stat(path)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return nil
}
return fmt.Errorf("read Cassandra configfile %s: %w", path, err)
}
if !info.Mode().IsRegular() {
return fmt.Errorf("Cassandra configfile is not a regular file: %s", path)
}
parsed, err := hocon.ParseResource(path)
if err != nil {
return fmt.Errorf("parse Cassandra configfile %s: %w", path, err)
}
config.configFile = path
if err := applyJavaDriverHOCON(config, parsed); err != nil {
return fmt.Errorf("apply Cassandra configfile %s: %w", path, err)
}
return nil
}
func applyJavaDriverHOCON(config *cassandraConfig, parsed *hocon.Config) error {
if value, ok, err := hoconDuration(parsed, javaDriverConfigPrefix+"basic.request.timeout"); err != nil {
return err
} else if ok {
config.requestTimeout = value
}
if value, ok, err := hoconString(parsed, javaDriverConfigPrefix+"basic.request.consistency"); err != nil {
return err
} else if ok {
if _, err := gocql.ParseConsistencyWrapper(value); err != nil {
return err
}
config.consistency = value
}
if value, ok, err := hoconString(parsed, javaDriverConfigPrefix+"basic.request.serial-consistency"); err != nil {
return err
} else if ok {
consistency, err := gocql.ParseConsistencyWrapper(value)
if err != nil {
return err
}
if consistency != gocql.Serial && consistency != gocql.LocalSerial {
return fmt.Errorf("serial consistency must be SERIAL or LOCAL_SERIAL")
}
config.serialConsistency = value
}
if value, ok, err := hoconInt(parsed, javaDriverConfigPrefix+"basic.request.page-size"); err != nil {
return err
} else if ok {
if value < 1 {
return fmt.Errorf("page size must be positive")
}
config.pageSize = value
}
if value, ok, err := hoconString(parsed, javaDriverConfigPrefix+"basic.load-balancing-policy.local-datacenter"); err != nil {
return err
} else if ok {
config.localDatacenter = value
}
if value, ok, err := hoconString(parsed, javaDriverConfigPrefix+"basic.load-balancing-policy.class"); err != nil {
return err
} else if ok {
policy, err := normalizeLoadBalancingPolicy(value)
if err != nil {
return err
}
config.loadBalancingPolicy = policy
}
if value, ok, err := hoconString(parsed, javaDriverConfigPrefix+"basic.cloud.secure-connect-bundle"); err != nil {
return err
} else if ok {
config.secureConnectBundle = value
}
if value, ok, err := hoconDuration(parsed, javaDriverConfigPrefix+"advanced.connection.connect-timeout"); err != nil {
return err
} else if ok {
config.connectTimeout = value
}
if value, ok, err := hoconInt(parsed, javaDriverConfigPrefix+"advanced.connection.pool.local.size"); err != nil {
return err
} else if ok {
if value < 1 || value > 32 {
return fmt.Errorf("connection pool local size must be between 1 and 32")
}
config.numConnections = value
}
if value, ok, err := hoconBool(parsed, javaDriverConfigPrefix+"advanced.socket.tcp-no-delay"); err != nil {
return err
} else if ok {
config.tcpNoDelay = value
}
if value, ok, err := hoconBool(parsed, javaDriverConfigPrefix+"advanced.socket.keep-alive"); err != nil {
return err
} else if ok {
config.keepAlive = value
}
if value, ok, err := hoconProtocolVersion(parsed, javaDriverConfigPrefix+"advanced.protocol.version"); err != nil {
return err
} else if ok {
config.protocolVersion = value
}
if value, ok, err := hoconString(parsed, javaDriverConfigPrefix+"advanced.retry-policy.class"); err != nil {
return err
} else if ok {
policy, err := normalizeRetryPolicy(value)
if err != nil {
return err
}
config.retryPolicy = policy
}
if value, ok, err := hoconString(parsed, javaDriverConfigPrefix+"advanced.reconnection-policy.class"); err != nil {
return err
} else if ok {
policy, baseDelay, maxDelay, err := parseReconnectionPolicy(value)
if err != nil {
return err
}
config.reconnectionPolicy = policy
config.reconnectionBaseDelay = baseDelay
config.reconnectionMaxDelay = maxDelay
}
if value, ok, err := hoconDuration(parsed, javaDriverConfigPrefix+"advanced.reconnection-policy.base-delay"); err != nil {
return err
} else if ok {
config.reconnectionBaseDelay = value
}
if value, ok, err := hoconDuration(parsed, javaDriverConfigPrefix+"advanced.reconnection-policy.max-delay"); err != nil {
return err
} else if ok {
config.reconnectionMaxDelay = value
}
if err := applyHOCONAuthentication(config, parsed); err != nil {
return err
}
if err := applyHOCONSSL(config, parsed); err != nil {
return err
}
return applyNativeHOCON(config, parsed)
}
func applyHOCONAuthentication(config *cassandraConfig, parsed *hocon.Config) error {
prefix := javaDriverConfigPrefix + "advanced.auth-provider."
if value, ok, err := hoconString(parsed, prefix+"class"); err != nil {
return err
} else if ok {
switch strings.ToLower(simpleClassName(value)) {
case "plaintextauthprovider", "dseplaintextauthprovider":
config.kerberos.enabled = false
case "kerberosauthprovider", "programmatickerberosauthprovider", "dsegssapiauthprovider":
config.kerberos.enabled = true
default:
return fmt.Errorf("unsupported Cassandra auth provider class: %s", value)
}
}
if value, ok, err := hoconString(parsed, prefix+"username"); err != nil {
return err
} else if ok {
config.username = value
}
if value, ok, err := hoconString(parsed, prefix+"password"); err != nil {
return err
} else if ok {
config.password = value
}
if value, ok, err := hoconString(parsed, prefix+"authorization-id"); err != nil {
return err
} else if ok {
config.kerberos.authorizationID = value
}
for _, path := range []string{prefix + "sasl-protocol", prefix + "service"} {
if value, ok, err := hoconString(parsed, path); err != nil {
return err
} else if ok {
config.kerberos.serviceName = value
}
}
if value, ok, err := hoconStringMap(parsed, prefix+"sasl-properties"); err != nil {
return err
} else if ok {
for key, property := range value {
if strings.EqualFold(key, "javax.security.sasl.qop") {
config.kerberos.qop = property
}
}
}
if value, ok, err := hoconString(parsed, prefix+"server-name-resolver"); err != nil {
return err
} else if ok && value != "" {
return fmt.Errorf("custom Java Kerberos server-name-resolver is not supported; use dbx.cassandra.kerberos.server-name")
}
loginPrefix := prefix + "login-configuration."
if value, ok, err := hoconString(parsed, loginPrefix+"principal"); err != nil {
return err
} else if ok {
config.kerberos.principal = value
}
if value, ok, err := firstHOCONString(parsed, loginPrefix+"keyTab", loginPrefix+"keytab"); err != nil {
return err
} else if ok {
config.kerberos.keytabPath = value
}
if value, ok, err := firstHOCONString(parsed, loginPrefix+"ticketCache", loginPrefix+"ticket-cache"); err != nil {
return err
} else if ok {
config.kerberos.ccachePath = value
}
if value, ok, err := firstHOCONBool(parsed, loginPrefix+"useKeyTab", loginPrefix+"use-keytab"); err != nil {
return err
} else if ok {
config.kerberos.useKeytab = value
config.kerberos.useKeytabSet = true
}
if value, ok, err := firstHOCONBool(parsed, loginPrefix+"useTicketCache", loginPrefix+"use-ticket-cache"); err != nil {
return err
} else if ok {
config.kerberos.useTicketCache = value
config.kerberos.useTicketCacheSet = true
}
return nil
}
func applyHOCONSSL(config *cassandraConfig, parsed *hocon.Config) error {
prefix := javaDriverConfigPrefix + "advanced.ssl-engine-factory."
if value, ok, err := hoconString(parsed, prefix+"class"); err != nil {
return err
} else if ok {
if !strings.EqualFold(simpleClassName(value), "DefaultSslEngineFactory") {
return fmt.Errorf("unsupported Cassandra SSL engine factory class: %s", value)
}
config.ssl = true
}
if value, ok, err := hoconBool(parsed, prefix+"hostname-validation"); err != nil {
return err
} else if ok {
config.hostVerification = value
config.ssl = true
}
for _, path := range []string{prefix + "truststore-path", prefix + "keystore-path"} {
if value, ok, err := hoconString(parsed, path); err != nil {
return err
} else if ok && value != "" {
return fmt.Errorf("Java truststore and keystore files are not supported; use dbx.cassandra.tls PEM paths")
}
}
return nil
}
func applyNativeHOCON(config *cassandraConfig, parsed *hocon.Config) error {
prefix := "dbx.cassandra."
stringMappings := []struct {
path string
target *string
}{
{"tls.ca-cert-path", &config.caCertPath},
{"tls.client-cert-path", &config.clientCertPath},
{"tls.client-key-path", &config.clientKeyPath},
{"kerberos.config", &config.kerberos.configPath},
{"kerberos.jaas-config", &config.kerberos.jaasConfigPath},
{"kerberos.principal", &config.kerberos.principal},
{"kerberos.realm", &config.kerberos.realm},
{"kerberos.keytab", &config.kerberos.keytabPath},
{"kerberos.ccache", &config.kerberos.ccachePath},
{"kerberos.password", &config.kerberos.password},
{"kerberos.service-name", &config.kerberos.serviceName},
{"kerberos.server-name", &config.kerberos.serverName},
{"kerberos.authorization-id", &config.kerberos.authorizationID},
{"kerberos.qop", &config.kerberos.qop},
}
for _, mapping := range stringMappings {
if value, ok, err := hoconString(parsed, prefix+mapping.path); err != nil {
return err
} else if ok {
*mapping.target = value
}
}
if value, ok, err := hoconBool(parsed, prefix+"tls.enabled"); err != nil {
return err
} else if ok {
config.ssl = value
}
if value, ok, err := hoconBool(parsed, prefix+"tls.hostname-verification"); err != nil {
return err
} else if ok {
config.hostVerification = value
}
if value, ok, err := hoconBool(parsed, prefix+"kerberos.enabled"); err != nil {
return err
} else if ok {
config.kerberos.enabled = value
}
if value, ok, err := hoconBool(parsed, prefix+"kerberos.disable-pafxfast"); err != nil {
return err
} else if ok {
config.kerberos.disablePAFXFAST = value
}
if value, ok, err := hoconBool(parsed, prefix+"kerberos.use-keytab"); err != nil {
return err
} else if ok {
config.kerberos.useKeytab = value
config.kerberos.useKeytabSet = true
}
if value, ok, err := hoconBool(parsed, prefix+"kerberos.use-ticket-cache"); err != nil {
return err
} else if ok {
config.kerberos.useTicketCache = value
config.kerberos.useTicketCacheSet = true
}
return nil
}
func normalizeLocalFilePath(raw string) (string, error) {
value := strings.TrimSpace(raw)
if value == "" {
return "", nil
}
if strings.Contains(value, "://") || strings.HasPrefix(strings.ToLower(value), "file:") {
parsed, err := url.Parse(value)
if err != nil {
return "", err
}
if parsed.Scheme != "file" {
return "", fmt.Errorf("unsupported file URI scheme: %s", parsed.Scheme)
}
if parsed.Host != "" && !strings.EqualFold(parsed.Host, "localhost") {
return "", fmt.Errorf("remote file URI hosts are not supported: %s", parsed.Host)
}
value, err = url.PathUnescape(parsed.Path)
if err != nil {
return "", err
}
if runtime.GOOS == "windows" && len(value) >= 3 && value[0] == '/' && value[2] == ':' {
value = value[1:]
}
}
return filepath.Clean(filepath.FromSlash(value)), nil
}
func hoconString(config *hocon.Config, path string) (string, bool, error) {
if config.Get(path) == nil {
return "", false, nil
}
value, err := config.GetStringE(path)
if err != nil {
return "", false, fmt.Errorf("invalid %s: %w", path, err)
}
return strings.TrimSpace(value), true, nil
}
func firstHOCONString(config *hocon.Config, paths ...string) (string, bool, error) {
for _, path := range paths {
value, ok, err := hoconString(config, path)
if err != nil || ok {
return value, ok, err
}
}
return "", false, nil
}
func hoconStringMap(config *hocon.Config, path string) (map[string]string, bool, error) {
if config.Get(path) == nil {
return nil, false, nil
}
value, err := config.GetStringMapStringE(path)
if err != nil {
return nil, false, fmt.Errorf("invalid %s: %w", path, err)
}
return value, true, nil
}
func hoconDuration(config *hocon.Config, path string) (time.Duration, bool, error) {
if config.Get(path) == nil {
return 0, false, nil
}
value, err := config.GetDurationE(path)
if err != nil {
return 0, false, fmt.Errorf("invalid %s: %w", path, err)
}
return value, true, nil
}
func hoconInt(config *hocon.Config, path string) (int, bool, error) {
if config.Get(path) == nil {
return 0, false, nil
}
value, err := config.GetIntE(path)
if err != nil {
return 0, false, fmt.Errorf("invalid %s: %w", path, err)
}
return value, true, nil
}
func hoconBool(config *hocon.Config, path string) (bool, bool, error) {
value := config.Get(path)
if value == nil {
return false, false, nil
}
switch typed := value.(type) {
case hocon.Boolean:
return bool(typed), true, nil
case hocon.String:
parsed, err := strconv.ParseBool(string(typed))
if err != nil {
return false, false, fmt.Errorf("invalid %s: %w", path, err)
}
return parsed, true, nil
default:
return false, false, fmt.Errorf("invalid %s: expected boolean", path)
}
}
func firstHOCONBool(config *hocon.Config, paths ...string) (bool, bool, error) {
for _, path := range paths {
value, ok, err := hoconBool(config, path)
if err != nil || ok {
return value, ok, err
}
}
return false, false, nil
}
func hoconProtocolVersion(config *hocon.Config, path string) (int, bool, error) {
value := config.Get(path)
if value == nil {
return 0, false, nil
}
var raw string
switch typed := value.(type) {
case hocon.Int:
raw = strconv.Itoa(int(typed))
case hocon.String:
raw = string(typed)
default:
return 0, false, fmt.Errorf("invalid %s: expected V3, V4, or V5", path)
}
raw = strings.TrimPrefix(strings.ToUpper(strings.TrimSpace(raw)), "V")
version, err := strconv.Atoi(raw)
if err != nil || version < 3 || version > 5 {
return 0, false, fmt.Errorf("invalid %s: expected V3, V4, or V5", path)
}
return version, true, nil
}
agents/drivers/cassandra-go/config_file_test.go
+173
View File
@@ -0,0 +1,173 @@
package main
import (
"net/url"
"os"
"path/filepath"
"reflect"
"testing"
"time"
"github.com/gurkankaymak/hocon"
)
func TestCassandraConfigFileOverridesURLExceptEndpoint(t *testing.T) {
configPath := writeTestFile(t, "application.conf", `
datastax-java-driver {
basic {
contact-points = ["ignored.example.com:9042"]
session-keyspace = ignored_keyspace
request {
timeout = 2 seconds
consistency = LOCAL_ONE
serial-consistency = LOCAL_SERIAL
page-size = 321
}
load-balancing-policy {
class = BasicLoadBalancingPolicy
local-datacenter = dc-config
}
}
advanced {
connection {
connect-timeout = 3 seconds
pool.local.size = 4
}
socket {
tcp-no-delay = false
keep-alive = true
}
protocol.version = V4
retry-policy.class = FallthroughRetryPolicy
reconnection-policy {
class = ExponentialReconnectionPolicy
base-delay = 4 seconds
max-delay = 20 seconds
}
auth-provider {
class = PlainTextAuthProvider
username = file-user
password = file-password
}
ssl-engine-factory {
class = DefaultSslEngineFactory
hostname-validation = false
}
}
}
`)
config, err := parseCassandraConfig(connectParams{
Host: "url.example.com",
Database: "url_keyspace",
Username: "url-user",
Password: "url-password",
URLParams: url.Values{
"configfile": []string{configPath},
"requesttimeout": []string{"30s"},
"connecttimeout": []string{"31s"},
"consistency": []string{"QUORUM"},
"serialconsistency": []string{"SERIAL"},
"fetchsize": []string{"999"},
"localdatacenter": []string{"dc-url"},
"protocolversion": []string{"5"},
"numconns": []string{"2"},
"tcpnodelay": []string{"true"},
"keepalive": []string{"false"},
"retry": []string{"DefaultRetryPolicy"},
"reconnection": []string{"ConstantReconnectionPolicy((long)1)"},
"user": []string{"query-user"},
"password": []string{"query-password"},
}.Encode(),
})
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(config.hosts, []string{"url.example.com"}) || config.keyspace != "url_keyspace" {
t.Fatalf("configfile must not replace endpoint or keyspace: %#v", config)
}
if config.username != "file-user" || config.password != "file-password" {
t.Fatalf("configfile credentials did not override URL values: %#v", config)
}
if config.requestTimeout != 2*time.Second || config.connectTimeout != 3*time.Second {
t.Fatalf("unexpected configfile timeouts: request=%s connect=%s", config.requestTimeout, config.connectTimeout)
}
if config.consistency != "LOCAL_ONE" || config.serialConsistency != "LOCAL_SERIAL" || config.pageSize != 321 {
t.Fatalf("unexpected request config: %#v", config)
}
if config.localDatacenter != "dc-config" || config.loadBalancingPolicy != "default" || config.protocolVersion != 4 {
t.Fatalf("unexpected topology/protocol config: %#v", config)
}
if config.numConnections != 4 || config.tcpNoDelay || !config.keepAlive {
t.Fatalf("unexpected connection/socket config: %#v", config)
}
if config.retryPolicy != "fallthrough" || config.reconnectionPolicy != "exponential" {
t.Fatalf("unexpected policy config: %#v", config)
}
if config.reconnectionBaseDelay != 4*time.Second || config.reconnectionMaxDelay != 20*time.Second {
t.Fatalf("unexpected reconnection delays: %#v", config)
}
if !config.ssl || config.hostVerification {
t.Fatalf("unexpected TLS config: %#v", config)
}
}
func TestCassandraConfigFileSupportsKerberosLoginOptionCasing(t *testing.T) {
parsed, err := hocon.ParseString(`
datastax-java-driver.advanced.auth-provider {
class = com.instaclustr.cassandra.driver.auth.KerberosAuthProvider
authorization-id = assumed_role
sasl-protocol = cassandra-custom
sasl-properties."javax.security.sasl.qop" = auth
login-configuration {
principal = "alice@EXAMPLE.COM"
keyTab = "/tmp/alice.keytab"
ticketCache = "FILE:/tmp/alice.ccache"
useKeyTab = true
useTicketCache = false
}
}
`)
if err != nil {
t.Fatal(err)
}
config := cassandraConfig{kerberos: defaultKerberosConfig()}
if err := applyJavaDriverHOCON(&config, parsed); err != nil {
t.Fatal(err)
}
if !config.kerberos.enabled || config.kerberos.principal != "alice@EXAMPLE.COM" {
t.Fatalf("unexpected Kerberos provider config: %#v", config.kerberos)
}
if config.kerberos.keytabPath != "/tmp/alice.keytab" || config.kerberos.ccachePath != "FILE:/tmp/alice.ccache" {
t.Fatalf("unexpected Kerberos file options: %#v", config.kerberos)
}
if !config.kerberos.useKeytab || !config.kerberos.useKeytabSet || config.kerberos.useTicketCache || !config.kerberos.useTicketCacheSet {
t.Fatalf("unexpected Kerberos credential switches: %#v", config.kerberos)
}
if config.kerberos.authorizationID != "assumed_role" || config.kerberos.serviceName != "cassandra-custom" || config.kerberos.qop != "auth" {
t.Fatalf("unexpected Kerberos SASL options: %#v", config.kerberos)
}
}
func TestMissingCassandraConfigFileIsIgnoredForJDBCCompatibility(t *testing.T) {
missingPath := filepath.Join(t.TempDir(), "missing.conf")
config, err := parseCassandraConfig(connectParams{
Host: "localhost",
URLParams: url.Values{"configfile": []string{missingPath}, "requesttimeout": []string{"2s"}}.Encode(),
})
if err != nil {
t.Fatal(err)
}
if config.requestTimeout != 2*time.Second {
t.Fatalf("missing configfile must leave URL options intact: %#v", config)
}
}
func writeTestFile(t *testing.T, name, contents string) string {
t.Helper()
path := filepath.Join(t.TempDir(), name)
if err := os.WriteFile(path, []byte(contents), 0o600); err != nil {
t.Fatal(err)
}
return path
}
agents/drivers/cassandra-go/config_test.go
+4 -5
View File
@@ -128,16 +128,15 @@ func TestParseCassandraConfigAcceptsDefaultSSLEngineFactory(t *testing.T) {
}
}
func TestParseCassandraConfigRejectsJavaOnlyJDBCOptions(t *testing.T) {
func TestParseCassandraConfigRejectsCustomJavaImplementationClasses(t *testing.T) {
tests := []string{
"configfile=/tmp/application.conf",
"secureconnectbundle=/tmp/secure-connect.zip",
"usekrb5=true",
"sslenginefactory=example.CustomSslEngineFactory",
"loadbalancing=example.CustomPolicy",
"retry=example.CustomRetryPolicy",
}
for _, urlParams := range tests {
if _, err := parseCassandraConfig(connectParams{Host: "localhost", URLParams: urlParams}); err == nil {
t.Fatalf("expected unsupported Java-only option error for %q", urlParams)
t.Fatalf("expected custom Java implementation rejection for %q", urlParams)
}
}
}
agents/drivers/cassandra-go/go.mod
+24 -2
View File
@@ -2,6 +2,28 @@ module github.com/t8y2/dbx/agents/drivers/cassandra-go
go 1.22
require github.com/apache/cassandra-gocql-driver/v2 v2.1.2
require (
github.com/apache/cassandra-gocql-driver/v2 v2.1.2
github.com/datastax/gocql-astra/v2 v2.0.1
github.com/gurkankaymak/hocon v1.3.0
github.com/jcmturner/gofork v1.7.6
github.com/jcmturner/gokrb5/v8 v8.4.4
)
require gopkg.in/inf.v0 v0.9.1 // indirect
require (
github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
github.com/datastax/astra-client-go/v2 v2.2.54 // indirect
github.com/datastax/cql-proxy v0.1.6 // indirect
github.com/datastax/go-cassandra-native-protocol v0.0.0-20220706104457-5e8aad05cf90 // indirect
github.com/deepmap/oapi-codegen v1.12.4 // indirect
github.com/google/uuid v1.3.0 // indirect
github.com/hashicorp/go-uuid v1.0.3 // indirect
github.com/jcmturner/aescts/v2 v2.0.0 // indirect
github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
github.com/jcmturner/rpc/v2 v2.0.3 // indirect
go.uber.org/multierr v1.10.0 // indirect
go.uber.org/zap v1.27.0 // indirect
golang.org/x/crypto v0.6.0 // indirect
golang.org/x/net v0.7.0 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
)
agents/drivers/cassandra-go/go.sum
+86
View File
@@ -1,9 +1,46 @@
github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
github.com/apache/cassandra-gocql-driver/v2 v2.1.2 h1:lu/p0Db2av18enHJvWJQoChLssI0P+AR06STq4VdvCc=
github.com/apache/cassandra-gocql-driver/v2 v2.1.2/go.mod h1:QH/asJjB3mHvY6Dot6ZKMMpTcOrWJ8i9GhsvG1g0PK4=
github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=
github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w=
github.com/datastax/astra-client-go/v2 v2.2.54 h1:R2k9ek9zaU15cLD96np5gsj12oZhK3Z5/tSytjQagO8=
github.com/datastax/astra-client-go/v2 v2.2.54/go.mod h1:zxXWuqDkYia7PzFIL3T7RmjChc9LN81UnfI2yB4kE7M=
github.com/datastax/cql-proxy v0.1.6 h1:IFJ/QV5Hk25CVaqVzPAz9o3ZsczZKKE3htpeNk3/e9o=
github.com/datastax/cql-proxy v0.1.6/go.mod h1:AgfO39+s+6oSwMETjysTnr5dhDArYfiDoMz0J9RG72I=
github.com/datastax/go-cassandra-native-protocol v0.0.0-20220706104457-5e8aad05cf90 h1:SiFe3gwoHPt95ly6HLjwyyItxROxCUJuxqqTnguR5ac=
github.com/datastax/go-cassandra-native-protocol v0.0.0-20220706104457-5e8aad05cf90/go.mod h1:6FzirJfdffakAVqmHjwVfFkpru/gNbIazUOK5rIhndc=
github.com/datastax/gocql-astra/v2 v2.0.1 h1:yZo7pN51mn/D8LlA7FXFrkYeEw2OrsbGcAZLfY9dx6M=
github.com/datastax/gocql-astra/v2 v2.0.1/go.mod h1:dnBksaFCHOIHWqVqDploY3/J8G+0PO/9193VBphW0cA=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/deepmap/oapi-codegen v1.12.4 h1:pPmn6qI9MuOtCz82WY2Xaw46EQjgvxednXXrP7g5Q2s=
github.com/deepmap/oapi-codegen v1.12.4/go.mod h1:3lgHGMu6myQ2vqbbTXH2H1o4eXFTGnFiDaOaKKl5yas=
github.com/golang/snappy v0.0.3 h1:fHPg5GQYlCeLIPB9BZqMVR5nR9A+IM5zcgeTdjMYmLA=
github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+HVt/4epWDjd4=
github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/zI+bUmuGM=
github.com/gurkankaymak/hocon v1.3.0 h1:OPw9XP+c3o4JGBVIS5e4+84YNfFFFgKQ625LX5tjSyQ=
github.com/gurkankaymak/hocon v1.3.0/go.mod h1:CM7yeDDq8AUU7I+QiLA7aVSJVbP1BVgZqqCiuzgcNS0=
github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
github.com/jcmturner/aescts/v2 v2.0.0/go.mod h1:AiaICIRyfYg35RUkr8yESTqvSy7csK90qZ5xfvvsoNs=
github.com/jcmturner/dnsutils/v2 v2.0.0 h1:lltnkeZGL0wILNvrNiVCR6Ro5PGU/SeBvVO/8c/iPbo=
github.com/jcmturner/dnsutils/v2 v2.0.0/go.mod h1:b0TnjGOvI/n42bZa+hmXL+kFJZsFT7G4t3HTlQ184QM=
github.com/jcmturner/gofork v1.7.6 h1:QH0l3hzAU1tfT3rZCnW5zXl+orbkNMMRGJfdJjHVETg=
github.com/jcmturner/gofork v1.7.6/go.mod h1:1622LH6i/EZqLloHfE7IeZ0uEJwMSUyQ/nDd82IeqRo=
github.com/jcmturner/goidentity/v6 v6.0.1 h1:VKnZd2oEIMorCTsFBnJWbExfNN7yZr3EhJAxwOkZg6o=
github.com/jcmturner/goidentity/v6 v6.0.1/go.mod h1:X1YW3bgtvwAXju7V3LCIMpY0Gbxyjn/mY9zx4tFonSg=
github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh687T8=
github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs=
github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY=
github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
@@ -14,15 +51,64 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY=
github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ=
github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
go.uber.org/atomic v1.8.0 h1:CUhrE4N1rqSE6FM9ecihEjRkLQu8cDfgDyoOs83mEY4=
go.uber.org/atomic v1.8.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.6.0 h1:qfktjS5LUO+fFKeJXZ+ikTRijMmljikvG68fpMMruSc=
golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.12.0 h1:CM0HF96J0hcLAwsHPJZjfdNzs0gftsLfgKt57wWHJ0o=
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
agents/drivers/cassandra-go/kerberos.go
+634
View File
@@ -0,0 +1,634 @@
package main
import (
"encoding/binary"
"fmt"
"net"
"os"
"os/user"
"path/filepath"
"regexp"
"runtime"
"strconv"
"strings"
gocql "github.com/apache/cassandra-gocql-driver/v2"
"github.com/jcmturner/gofork/encoding/asn1"
"github.com/jcmturner/gokrb5/v8/asn1tools"
krb5client "github.com/jcmturner/gokrb5/v8/client"
krb5config "github.com/jcmturner/gokrb5/v8/config"
"github.com/jcmturner/gokrb5/v8/credentials"
"github.com/jcmturner/gokrb5/v8/gssapi"
"github.com/jcmturner/gokrb5/v8/iana/chksumtype"
"github.com/jcmturner/gokrb5/v8/iana/keyusage"
"github.com/jcmturner/gokrb5/v8/keytab"
"github.com/jcmturner/gokrb5/v8/messages"
"github.com/jcmturner/gokrb5/v8/types"
)
const (
kerberosAPRequestTokenID = 0x0100
kerberosGSSAPITag = 0x60
kerberosSecurityNone = 0x01
)
type kerberosCredentialMode int
const (
kerberosCredentialNone kerberosCredentialMode = iota
kerberosCredentialPassword
kerberosCredentialKeytab
kerberosCredentialCCache
)
type kerberosConfig struct {
enabled bool
configPath string
jaasConfigPath string
principal string
realm string
keytabPath string
ccachePath string
password string
serviceName string
serverName string
authorizationID string
qop string
disablePAFXFAST bool
useKeytab bool
useKeytabSet bool
useTicketCache bool
useTicketCacheSet bool
credentialMode kerberosCredentialMode
credentialUser string
credentialRealm string
}
type kerberosAuthenticator struct {
domain string
clientName types.PrincipalName
ticket messages.Ticket
sessionKey types.EncryptionKey
authorizationID string
step int
}
var (
jaasBlockPattern = regexp.MustCompile(`(?is)\bCassandraJavaClient\s*\{(.*?)\}\s*;`)
jaasModulePattern = regexp.MustCompile(`(?is)\bcom\.sun\.security\.auth\.module\.Krb5LoginModule\b(.*?);`)
jaasOptionPattern = regexp.MustCompile(`(?is)([A-Za-z][A-Za-z0-9_-]*)\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\s;]+))`)
)
func defaultKerberosConfig() kerberosConfig {
return kerberosConfig{
serviceName: "cassandra",
qop: "auth",
}
}
func (config *kerberosConfig) finalize(username, password string) error {
config.applyJavaSystemProperties()
if config.jaasConfigPath != "" {
path, err := normalizeLocalFilePath(config.jaasConfigPath)
if err != nil {
return fmt.Errorf("invalid Cassandra JAAS config path: %w", err)
}
config.jaasConfigPath = path
if err := config.applyJAASConfig(path); err != nil {
return err
}
}
config.applyKerberosConfigEnvironment()
if config.configPath == "" {
config.configPath = defaultKerberosConfigPath()
}
path, err := normalizeLocalFilePath(firstPathListEntry(config.configPath))
if err != nil {
return fmt.Errorf("invalid Kerberos config path: %w", err)
}
config.configPath = path
if err := requireRegularFile("Kerberos config", config.configPath); err != nil {
return err
}
krbConfig, err := krb5config.Load(config.configPath)
if err != nil {
return fmt.Errorf("load Kerberos config %s: %w", config.configPath, err)
}
if config.serviceName == "" {
config.serviceName = "cassandra"
}
if !kerberosQOPIncludesAuth(config.qop) {
return fmt.Errorf("Cassandra Kerberos currently supports SASL QOP auth only, got %s", config.qop)
}
config.qop = "auth"
if config.principal == "" {
config.principal = strings.TrimSpace(username)
}
if config.password == "" {
config.password = password
}
if config.useTicketCache {
return config.selectCCacheCredential()
}
if config.useKeytab {
return config.selectKeytabCredential(krbConfig)
}
if config.ccachePath != "" && !config.useTicketCacheSet {
return config.selectCCacheCredential()
}
if config.keytabPath != "" && !config.useKeytabSet {
return config.selectKeytabCredential(krbConfig)
}
if config.principal != "" && config.password != "" {
config.credentialUser, config.credentialRealm, err = splitKerberosPrincipal(
config.principal,
config.realm,
krbConfig.LibDefaults.DefaultRealm,
)
if err != nil {
return err
}
config.credentialMode = kerberosCredentialPassword
return nil
}
if !config.useTicketCacheSet {
config.ccachePath = os.Getenv("KRB5CCNAME")
if config.ccachePath == "" {
defaultCache := defaultKerberosCCachePath()
if path, normalizeErr := normalizeKerberosCachePath(defaultCache); normalizeErr == nil {
if info, statErr := os.Stat(path); statErr == nil && info.Mode().IsRegular() {
config.ccachePath = defaultCache
}
}
}
if config.ccachePath != "" {
return config.selectCCacheCredential()
}
}
if !config.useKeytabSet {
config.keytabPath = firstNonEmpty(os.Getenv("KRB5_CLIENT_KTNAME"), os.Getenv("KRB5_KTNAME"))
if config.keytabPath != "" {
return config.selectKeytabCredential(krbConfig)
}
}
return fmt.Errorf("Kerberos authentication requires a credential cache, keytab, or principal and password")
}
func (config *kerberosConfig) selectCCacheCredential() error {
var err error
if config.ccachePath == "" {
config.ccachePath = defaultKerberosCCachePath()
}
config.ccachePath, err = normalizeKerberosCachePath(config.ccachePath)
if err != nil {
return err
}
if err := requireRegularFile("Kerberos credential cache", config.ccachePath); err != nil {
return err
}
config.credentialMode = kerberosCredentialCCache
return nil
}
func (config *kerberosConfig) selectKeytabCredential(krbConfig *krb5config.Config) error {
var err error
if config.keytabPath == "" {
config.keytabPath = firstNonEmpty(os.Getenv("KRB5_CLIENT_KTNAME"), os.Getenv("KRB5_KTNAME"))
if config.keytabPath == "" {
return fmt.Errorf("Kerberos keytab authentication requires a keytab path")
}
}
config.keytabPath, err = normalizeKerberosFileReference(config.keytabPath)
if err != nil {
return err
}
if err := requireRegularFile("Kerberos keytab", config.keytabPath); err != nil {
return err
}
if config.principal == "" {
config.principal, err = principalFromKeytab(config.keytabPath)
if err != nil {
return err
}
}
config.credentialUser, config.credentialRealm, err = splitKerberosPrincipal(
config.principal,
config.realm,
krbConfig.LibDefaults.DefaultRealm,
)
if err != nil {
return err
}
config.credentialMode = kerberosCredentialKeytab
return nil
}
func newKerberosAuthProvider(
config kerberosConfig,
username string,
password string,
) (func(*gocql.HostInfo) (gocql.Authenticator, error), error) {
if !config.enabled {
return nil, fmt.Errorf("Kerberos authentication is not enabled")
}
if config.credentialMode == kerberosCredentialNone {
if err := config.finalize(username, password); err != nil {
return nil, err
}
}
krbConfig, err := krb5config.Load(config.configPath)
if err != nil {
return nil, fmt.Errorf("load Kerberos config %s: %w", config.configPath, err)
}
return func(host *gocql.HostInfo) (gocql.Authenticator, error) {
return newKerberosAuthenticator(config, krbConfig, host)
}, nil
}
func newKerberosAuthenticator(
config kerberosConfig,
krbConfig *krb5config.Config,
host *gocql.HostInfo,
) (gocql.Authenticator, error) {
client, err := newKerberosClient(config, krbConfig)
if err != nil {
return nil, err
}
if err := client.Login(); err != nil {
client.Destroy()
return nil, fmt.Errorf("Kerberos login failed: %w", err)
}
serverName, err := kerberosServerName(config, host)
if err != nil {
client.Destroy()
return nil, err
}
servicePrincipal := config.serviceName + "/" + serverName
ticket, sessionKey, err := client.GetServiceTicket(servicePrincipal)
if err != nil {
client.Destroy()
return nil, fmt.Errorf("get Kerberos service ticket for %s: %w", servicePrincipal, err)
}
clientName := client.Credentials.CName()
clientName.NameString = append([]string(nil), clientName.NameString...)
authenticator := &kerberosAuthenticator{
domain: strings.Clone(client.Credentials.Domain()),
clientName: clientName,
ticket: ticket,
sessionKey: sessionKey,
authorizationID: config.authorizationID,
}
client.Destroy()
return authenticator, nil
}
func newKerberosClient(config kerberosConfig, krbConfig *krb5config.Config) (*krb5client.Client, error) {
settings := []func(*krb5client.Settings){krb5client.DisablePAFXFAST(config.disablePAFXFAST)}
switch config.credentialMode {
case kerberosCredentialCCache:
cache, err := credentials.LoadCCache(config.ccachePath)
if err != nil {
return nil, fmt.Errorf("load Kerberos credential cache %s: %w", config.ccachePath, err)
}
client, err := krb5client.NewFromCCache(cache, krbConfig, settings...)
if err != nil {
return nil, fmt.Errorf("create Kerberos client from credential cache: %w", err)
}
return client, nil
case kerberosCredentialKeytab:
loadedKeytab, err := keytab.Load(config.keytabPath)
if err != nil {
return nil, fmt.Errorf("load Kerberos keytab %s: %w", config.keytabPath, err)
}
return krb5client.NewWithKeytab(
config.credentialUser,
config.credentialRealm,
loadedKeytab,
krbConfig,
settings...,
), nil
case kerberosCredentialPassword:
return krb5client.NewWithPassword(
config.credentialUser,
config.credentialRealm,
config.password,
krbConfig,
settings...,
), nil
default:
return nil, fmt.Errorf("Kerberos credentials are not configured")
}
}
func (authenticator *kerberosAuthenticator) Challenge(request []byte) ([]byte, gocql.Authenticator, error) {
switch authenticator.step {
case 0:
token, err := authenticator.initialToken()
if err != nil {
return nil, nil, err
}
authenticator.step = 1
return token, authenticator, nil
case 1:
token, err := authenticator.securityLayerResponse(request)
if err != nil {
return nil, nil, err
}
authenticator.step = 2
return token, authenticator, nil
default:
return nil, nil, fmt.Errorf("unexpected Cassandra Kerberos authentication challenge")
}
}
func (authenticator *kerberosAuthenticator) Success(_ []byte) error {
if authenticator.step != 2 {
return fmt.Errorf("Cassandra reported Kerberos success before SASL negotiation completed")
}
return nil
}
func (authenticator *kerberosAuthenticator) initialToken() ([]byte, error) {
value, err := types.NewAuthenticator(authenticator.domain, authenticator.clientName)
if err != nil {
return nil, err
}
value.Cksum = types.Checksum{
CksumType: chksumtype.GSSAPI,
Checksum: kerberosAuthenticatorChecksum(),
}
request, err := messages.NewAPReq(authenticator.ticket, authenticator.sessionKey, value)
if err != nil {
return nil, err
}
payload := make([]byte, 2)
binary.BigEndian.PutUint16(payload, kerberosAPRequestTokenID)
encodedRequest, err := request.Marshal()
if err != nil {
return nil, err
}
payload = append(payload, encodedRequest...)
encodedOID, err := asn1.Marshal(gssapi.OIDKRB5.OID())
if err != nil {
return nil, err
}
header := append([]byte{kerberosGSSAPITag}, asn1tools.MarshalLengthBytes(len(encodedOID)+len(payload))...)
header = append(header, encodedOID...)
return append(header, payload...), nil
}
func (authenticator *kerberosAuthenticator) securityLayerResponse(challenge []byte) ([]byte, error) {
var request gssapi.WrapToken
if err := request.Unmarshal(challenge, true); err != nil {
return nil, fmt.Errorf("decode Kerberos SASL security-layer challenge: %w", err)
}
valid, err := request.Verify(authenticator.sessionKey, keyusage.GSSAPI_ACCEPTOR_SEAL)
if err != nil {
return nil, fmt.Errorf("verify Kerberos SASL security-layer challenge: %w", err)
}
if !valid {
return nil, fmt.Errorf("invalid Kerberos SASL security-layer challenge")
}
if len(request.Payload) < 4 || request.Payload[0]&kerberosSecurityNone == 0 {
return nil, fmt.Errorf("Cassandra Kerberos server does not allow SASL QOP auth")
}
payload := []byte{kerberosSecurityNone, 0, 0, 0}
payload = append(payload, authenticator.authorizationID...)
response, err := gssapi.NewInitiatorWrapToken(payload, authenticator.sessionKey)
if err != nil {
return nil, err
}
return response.Marshal()
}
func kerberosAuthenticatorChecksum() []byte {
checksum := make([]byte, 24)
binary.LittleEndian.PutUint32(checksum[:4], 16)
flags := uint32(gssapi.ContextFlagInteg | gssapi.ContextFlagConf)
binary.LittleEndian.PutUint32(checksum[20:24], flags)
return checksum
}
func kerberosQOPIncludesAuth(value string) bool {
for _, qop := range strings.Split(value, ",") {
if strings.EqualFold(strings.TrimSpace(qop), "auth") {
return true
}
}
return false
}
func kerberosServerName(config kerberosConfig, host *gocql.HostInfo) (string, error) {
if config.serverName != "" {
return strings.TrimSuffix(strings.TrimSpace(config.serverName), "."), nil
}
if host == nil {
return "", fmt.Errorf("resolve Kerberos server name: Cassandra host is unavailable")
}
address := host.ConnectAddress()
if address != nil {
names, err := net.LookupAddr(address.String())
if err == nil && len(names) > 0 {
return strings.TrimSuffix(strings.TrimSpace(names[0]), "."), nil
}
}
hostname, _, err := net.SplitHostPort(host.HostnameAndPort())
if err == nil && hostname != "" && net.ParseIP(hostname) == nil {
return strings.TrimSuffix(hostname, "."), nil
}
return "", fmt.Errorf("resolve Kerberos server name for Cassandra host %s; configure kerberosservername explicitly", host.ConnectAddressAndPort())
}
func (config *kerberosConfig) applyJavaSystemProperties() {
if config.jaasConfigPath == "" {
config.jaasConfigPath = javaSystemProperty("java.security.auth.login.config")
}
if config.configPath == "" {
config.configPath = javaSystemProperty("java.security.krb5.conf")
}
}
func (config *kerberosConfig) applyKerberosConfigEnvironment() {
if config.configPath == "" {
config.configPath = os.Getenv("KRB5_CONFIG")
}
}
func (config *kerberosConfig) applyJAASConfig(path string) error {
contents, err := os.ReadFile(path)
if err != nil {
return fmt.Errorf("read Cassandra JAAS config %s: %w", path, err)
}
block := jaasBlockPattern.FindSubmatch(contents)
if len(block) != 2 {
return fmt.Errorf("Cassandra JAAS config %s does not contain CassandraJavaClient", path)
}
module := jaasModulePattern.FindSubmatch(block[1])
if len(module) != 2 {
return fmt.Errorf("CassandraJavaClient in %s does not configure Krb5LoginModule", path)
}
options := map[string]string{}
for _, match := range jaasOptionPattern.FindAllSubmatch(module[1], -1) {
value := firstNonEmpty(string(match[2]), string(match[3]), string(match[4]))
options[strings.ToLower(string(match[1]))] = value
}
if config.principal == "" {
config.principal = options["principal"]
}
if config.keytabPath == "" {
config.keytabPath = options["keytab"]
}
if config.ccachePath == "" {
config.ccachePath = options["ticketcache"]
}
if value, ok := options["usekeytab"]; ok {
config.useKeytab, err = strconv.ParseBool(value)
if err != nil {
return fmt.Errorf("invalid useKeyTab in Cassandra JAAS config: %w", err)
}
config.useKeytabSet = true
}
if value, ok := options["useticketcache"]; ok {
config.useTicketCache, err = strconv.ParseBool(value)
if err != nil {
return fmt.Errorf("invalid useTicketCache in Cassandra JAAS config: %w", err)
}
config.useTicketCacheSet = true
}
return nil
}
func javaSystemProperty(name string) string {
pattern := regexp.MustCompile(`(?:^|\s)-D` + regexp.QuoteMeta(name) + `=(?:"([^"]*)"|'([^']*)'|(\S+))`)
for _, environmentName := range []string{"JAVA_TOOL_OPTIONS", "_JAVA_OPTIONS", "JDK_JAVA_OPTIONS"} {
match := pattern.FindStringSubmatch(os.Getenv(environmentName))
if len(match) == 4 {
return firstNonEmpty(match[1], match[2], match[3])
}
}
return ""
}
func normalizeKerberosCachePath(raw string) (string, error) {
value := strings.TrimSpace(raw)
if value == "" {
return "", fmt.Errorf("Kerberos ticket cache path is empty")
}
if separator := strings.IndexByte(value, ':'); separator > 0 && !isWindowsDrivePath(value) {
cacheType := strings.ToUpper(value[:separator])
if cacheType != "FILE" {
return "", fmt.Errorf("Kerberos credential cache type %s is not supported; use a FILE cache or keytab", cacheType)
}
value = value[separator+1:]
}
path, err := normalizeLocalFilePath(value)
if err != nil {
return "", fmt.Errorf("invalid Kerberos credential cache path: %w", err)
}
return path, nil
}
func isWindowsDrivePath(value string) bool {
return len(value) >= 3 && ((value[0] >= 'A' && value[0] <= 'Z') || (value[0] >= 'a' && value[0] <= 'z')) &&
value[1] == ':' && (value[2] == '\\' || value[2] == '/')
}
func normalizeKerberosFileReference(raw string) (string, error) {
value := strings.TrimSpace(raw)
if strings.HasPrefix(strings.ToUpper(value), "FILE:") {
value = value[5:]
}
path, err := normalizeLocalFilePath(value)
if err != nil {
return "", fmt.Errorf("invalid Kerberos file path: %w", err)
}
return path, nil
}
func splitKerberosPrincipal(principal, configuredRealm, defaultRealm string) (string, string, error) {
value := strings.TrimSpace(principal)
realm := strings.TrimSpace(configuredRealm)
if separator := strings.LastIndexByte(value, '@'); separator >= 0 {
realm = value[separator+1:]
value = value[:separator]
}
if value == "" {
return "", "", fmt.Errorf("Kerberos principal is empty")
}
if realm == "" {
realm = strings.TrimSpace(defaultRealm)
}
if realm == "" {
return "", "", fmt.Errorf("Kerberos realm is required for principal %s", principal)
}
return value, realm, nil
}
func principalFromKeytab(path string) (string, error) {
loadedKeytab, err := keytab.Load(path)
if err != nil {
return "", fmt.Errorf("load Kerberos keytab %s: %w", path, err)
}
principals := map[string]struct{}{}
for _, entry := range loadedKeytab.Entries {
principals[entry.Principal.String()] = struct{}{}
}
if len(principals) != 1 {
return "", fmt.Errorf("Kerberos keytab %s contains %d principals; configure kerberosprincipal explicitly", path, len(principals))
}
for principal := range principals {
return principal, nil
}
return "", fmt.Errorf("Kerberos keytab %s contains no principals", path)
}
func defaultKerberosConfigPath() string {
if runtime.GOOS == "windows" {
if windowsDirectory := os.Getenv("WINDIR"); windowsDirectory != "" {
return filepath.Join(windowsDirectory, "krb5.ini")
}
}
return "/etc/krb5.conf"
}
func defaultKerberosCCachePath() string {
if value := os.Getenv("KRB5CCNAME"); value != "" {
return value
}
currentUser, err := user.Current()
if err == nil && currentUser.Uid != "" {
return filepath.Join(os.TempDir(), "krb5cc_"+currentUser.Uid)
}
return ""
}
func firstPathListEntry(value string) string {
entries := filepath.SplitList(value)
if len(entries) == 0 {
return value
}
return entries[0]
}
func requireRegularFile(label, path string) error {
if path == "" {
return fmt.Errorf("%s path is empty", label)
}
info, err := os.Stat(path)
if err != nil {
return fmt.Errorf("read %s %s: %w", label, path, err)
}
if !info.Mode().IsRegular() {
return fmt.Errorf("%s is not a regular file: %s", label, path)
}
return nil
}
func firstNonEmpty(values ...string) string {
for _, value := range values {
if value != "" {
return value
}
}
return ""
}
agents/drivers/cassandra-go/kerberos_test.go
+307
View File
@@ -0,0 +1,307 @@
package main
import (
"bytes"
"encoding/binary"
"net"
"path/filepath"
"strings"
"testing"
gocql "github.com/apache/cassandra-gocql-driver/v2"
"github.com/jcmturner/gofork/encoding/asn1"
"github.com/jcmturner/gokrb5/v8/gssapi"
"github.com/jcmturner/gokrb5/v8/iana/keyusage"
"github.com/jcmturner/gokrb5/v8/messages"
"github.com/jcmturner/gokrb5/v8/types"
)
func TestKerberosPasswordCredentialsTakePrecedenceOverEnvironmentCache(t *testing.T) {
clearKerberosEnvironment(t)
t.Setenv("KRB5CCNAME", filepath.Join(t.TempDir(), "missing.ccache"))
config := defaultKerberosConfig()
config.enabled = true
config.configPath = writeKerberosConfig(t)
config.principal = "alice@EXAMPLE.COM"
config.password = "secret"
if err := config.finalize("", ""); err != nil {
t.Fatal(err)
}
if config.credentialMode != kerberosCredentialPassword || config.credentialUser != "alice" || config.credentialRealm != "EXAMPLE.COM" {
t.Fatalf("unexpected password credential selection: %#v", config)
}
}
func TestKerberosExplicitCredentialSourcesTakePrecedence(t *testing.T) {
clearKerberosEnvironment(t)
krb5Path := writeKerberosConfig(t)
cachePath := writeTestFile(t, "alice.ccache", "placeholder")
keytabPath := writeTestFile(t, "alice.keytab", "placeholder")
t.Run("ccache", func(t *testing.T) {
config := defaultKerberosConfig()
config.enabled = true
config.configPath = krb5Path
config.ccachePath = "FILE:" + cachePath
config.principal = "alice@EXAMPLE.COM"
config.password = "ignored"
if err := config.finalize("", ""); err != nil {
t.Fatal(err)
}
if config.credentialMode != kerberosCredentialCCache || config.ccachePath != cachePath {
t.Fatalf("unexpected ccache credential selection: %#v", config)
}
})
t.Run("keytab", func(t *testing.T) {
config := defaultKerberosConfig()
config.enabled = true
config.configPath = krb5Path
config.keytabPath = "FILE:" + keytabPath
config.principal = "alice@EXAMPLE.COM"
config.password = "ignored"
if err := config.finalize("", ""); err != nil {
t.Fatal(err)
}
if config.credentialMode != kerberosCredentialKeytab || config.keytabPath != keytabPath {
t.Fatalf("unexpected keytab credential selection: %#v", config)
}
})
}
func TestKerberosDiscoversJavaJAASAndKrb5Properties(t *testing.T) {
clearKerberosEnvironment(t)
krb5Path := writeKerberosConfig(t)
cachePath := writeTestFile(t, "alice.ccache", "placeholder")
jaasPath := writeTestFile(t, "jaas.conf", `
CassandraJavaClient {
com.sun.security.auth.module.Krb5LoginModule required
useTicketCache=true
ticketCache="FILE:`+cachePath+`"
principal="alice@EXAMPLE.COM";
};
`)
t.Setenv("JAVA_TOOL_OPTIONS", `-Djava.security.krb5.conf="`+krb5Path+`" -Djava.security.auth.login.config='`+jaasPath+`'`)
config := defaultKerberosConfig()
config.enabled = true
if err := config.finalize("", ""); err != nil {
t.Fatal(err)
}
if config.configPath != krb5Path || config.jaasConfigPath != jaasPath {
t.Fatalf("Java system properties were not applied: %#v", config)
}
if config.credentialMode != kerberosCredentialCCache || config.ccachePath != cachePath || config.principal != "alice@EXAMPLE.COM" {
t.Fatalf("JAAS credential cache was not applied: %#v", config)
}
}
func TestKerberosRejectsUnsupportedQOP(t *testing.T) {
clearKerberosEnvironment(t)
config := defaultKerberosConfig()
config.enabled = true
config.configPath = writeKerberosConfig(t)
config.principal = "alice@EXAMPLE.COM"
config.password = "secret"
config.qop = "auth-conf"
err := config.finalize("", "")
if err == nil || !strings.Contains(err.Error(), "supports SASL QOP auth only") {
t.Fatalf("expected QOP rejection, got %v", err)
}
}
func TestKerberosAcceptsQOPPreferenceListContainingAuth(t *testing.T) {
clearKerberosEnvironment(t)
config := defaultKerberosConfig()
config.enabled = true
config.configPath = writeKerberosConfig(t)
config.principal = "alice@EXAMPLE.COM"
config.password = "secret"
config.qop = "auth-conf, auth"
if err := config.finalize("", ""); err != nil {
t.Fatal(err)
}
if config.qop != "auth" {
t.Fatalf("unexpected negotiated QOP preference: %q", config.qop)
}
}
func TestNormalizeKerberosCachePathSupportsWindowsDrivePaths(t *testing.T) {
path, err := normalizeKerberosCachePath(`C:\Users\alice\krb5cc`)
if err != nil {
t.Fatal(err)
}
if path != `C:\Users\alice\krb5cc` {
t.Fatalf("unexpected Windows cache path: %q", path)
}
if _, err := normalizeKerberosCachePath("DIR:/tmp/krb5cc"); err == nil {
t.Fatal("expected non-FILE credential cache type rejection")
}
}
func TestKerberosServerNameSupportsExplicitOverride(t *testing.T) {
name, err := kerberosServerName(kerberosConfig{serverName: "node1.example.com."}, nil)
if err != nil {
t.Fatal(err)
}
if name != "node1.example.com" {
t.Fatalf("unexpected explicit server name: %q", name)
}
if _, err := kerberosServerName(kerberosConfig{}, nil); err == nil {
t.Fatal("expected missing host error")
}
host, err := gocql.NewHostInfoFromAddrPort(net.ParseIP("127.0.0.1"), 9042)
if err != nil {
t.Fatal(err)
}
if resolved, err := kerberosServerName(kerberosConfig{}, host); err != nil || strings.TrimSpace(resolved) == "" {
t.Fatalf("expected loopback canonical name, got %q, %v", resolved, err)
}
}
func TestKerberosInitialTokenContainsDecryptableAPRequest(t *testing.T) {
key := testKerberosEncryptionKey()
authenticator := kerberosAuthenticator{
domain: "EXAMPLE.COM",
clientName: types.NewPrincipalName(1, "alice"),
ticket: messages.Ticket{
TktVNO: 5,
Realm: "EXAMPLE.COM",
SName: types.NewPrincipalName(2, "cassandra/node1.example.com"),
EncPart: types.EncryptedData{
EType: key.KeyType,
KVNO: 1,
Cipher: []byte{1},
},
},
sessionKey: key,
}
token, err := authenticator.initialToken()
if err != nil {
t.Fatal(err)
}
if len(token) < 2 || token[0] != kerberosGSSAPITag {
t.Fatalf("unexpected GSSAPI token prefix: %x", token)
}
encodedOID, err := asn1.Marshal(gssapi.OIDKRB5.OID())
if err != nil {
t.Fatal(err)
}
oidOffset := bytes.Index(token, encodedOID)
if oidOffset < 0 {
t.Fatalf("Kerberos OID missing from token: %x", token)
}
payload := token[oidOffset+len(encodedOID):]
if len(payload) < 3 || binary.BigEndian.Uint16(payload[:2]) != kerberosAPRequestTokenID {
t.Fatalf("unexpected Kerberos mechanism token: %x", payload)
}
var request messages.APReq
if err := request.Unmarshal(payload[2:]); err != nil {
t.Fatal(err)
}
if err := request.DecryptAuthenticator(key); err != nil {
t.Fatal(err)
}
if request.Authenticator.CName.PrincipalNameString() != "alice" || request.Authenticator.CRealm != "EXAMPLE.COM" {
t.Fatalf("unexpected AP-REQ authenticator: %#v", request.Authenticator)
}
}
func TestKerberosSecurityLayerNegotiatesAuthAndAuthorizationID(t *testing.T) {
key := testKerberosEncryptionKey()
challenge := marshalKerberosAcceptorToken(t, key, []byte{0x07, 0x00, 0x10, 0x00})
authenticator := kerberosAuthenticator{sessionKey: key, authorizationID: "assumed_role"}
response, err := authenticator.securityLayerResponse(challenge)
if err != nil {
t.Fatal(err)
}
var decoded gssapi.WrapToken
if err := decoded.Unmarshal(response, false); err != nil {
t.Fatal(err)
}
valid, err := decoded.Verify(key, keyusage.GSSAPI_INITIATOR_SEAL)
if err != nil || !valid {
t.Fatalf("invalid security-layer response: valid=%t err=%v", valid, err)
}
want := append([]byte{kerberosSecurityNone, 0, 0, 0}, []byte("assumed_role")...)
if !bytes.Equal(decoded.Payload, want) {
t.Fatalf("unexpected security-layer payload: %x", decoded.Payload)
}
}
func TestKerberosSecurityLayerRejectsUnavailableAuthQOP(t *testing.T) {
key := testKerberosEncryptionKey()
challenge := marshalKerberosAcceptorToken(t, key, []byte{0x02, 0, 0, 0})
authenticator := kerberosAuthenticator{sessionKey: key}
if _, err := authenticator.securityLayerResponse(challenge); err == nil {
t.Fatal("expected server QOP rejection")
}
}
func TestKerberosAuthenticatorChecksumRequestsIntegrityAndConfidentiality(t *testing.T) {
checksum := kerberosAuthenticatorChecksum()
if len(checksum) != 24 || binary.LittleEndian.Uint32(checksum[:4]) != 16 {
t.Fatalf("unexpected channel-binding checksum: %x", checksum)
}
wantFlags := uint32(gssapi.ContextFlagInteg | gssapi.ContextFlagConf)
if flags := binary.LittleEndian.Uint32(checksum[20:24]); flags != wantFlags {
t.Fatalf("unexpected GSSAPI context flags: %x", flags)
}
}
func clearKerberosEnvironment(t *testing.T) {
t.Helper()
for _, name := range []string{
"JAVA_TOOL_OPTIONS",
"_JAVA_OPTIONS",
"JDK_JAVA_OPTIONS",
"KRB5_CONFIG",
"KRB5CCNAME",
"KRB5_CLIENT_KTNAME",
"KRB5_KTNAME",
} {
t.Setenv(name, "")
}
}
func writeKerberosConfig(t *testing.T) string {
t.Helper()
return writeTestFile(t, "krb5.conf", `
[libdefaults]
default_realm = EXAMPLE.COM
dns_lookup_realm = false
dns_lookup_kdc = false
[realms]
EXAMPLE.COM = {
kdc = 127.0.0.1:88
}
`)
}
func testKerberosEncryptionKey() types.EncryptionKey {
return types.EncryptionKey{KeyType: 18, KeyValue: bytes.Repeat([]byte{0x42}, 32)}
}
func marshalKerberosAcceptorToken(t *testing.T, key types.EncryptionKey, payload []byte) []byte {
t.Helper()
token := gssapi.WrapToken{
Flags: 0x01,
EC: 12,
Payload: payload,
}
if err := token.SetCheckSum(key, keyusage.GSSAPI_ACCEPTOR_SEAL); err != nil {
t.Fatal(err)
}
encoded, err := token.Marshal()
if err != nil {
t.Fatal(err)
}
return encoded
}