diff --git a/agents/drivers/cassandra-go/README.md b/agents/drivers/cassandra-go/README.md index b87b89ea7d..32cf684988 100644 --- a/agents/drivers/cassandra-go/README.md +++ b/agents/drivers/cassandra-go/README.md @@ -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/`. 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 diff --git a/agents/drivers/cassandra-go/astra_test.go b/agents/drivers/cassandra-go/astra_test.go new file mode 100644 index 0000000000..6c8020c9f7 --- /dev/null +++ b/agents/drivers/cassandra-go/astra_test.go @@ -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 +} diff --git a/agents/drivers/cassandra-go/config.go b/agents/drivers/cassandra-go/config.go index 462ae3299a..39bd11d2e0 100644 --- a/agents/drivers/cassandra-go/config.go +++ b/agents/drivers/cassandra-go/config.go @@ -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 diff --git a/agents/drivers/cassandra-go/config_file.go b/agents/drivers/cassandra-go/config_file.go new file mode 100644 index 0000000000..d20b0d7a69 --- /dev/null +++ b/agents/drivers/cassandra-go/config_file.go @@ -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 +} diff --git a/agents/drivers/cassandra-go/config_file_test.go b/agents/drivers/cassandra-go/config_file_test.go new file mode 100644 index 0000000000..ae81f80cbe --- /dev/null +++ b/agents/drivers/cassandra-go/config_file_test.go @@ -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 +} diff --git a/agents/drivers/cassandra-go/config_test.go b/agents/drivers/cassandra-go/config_test.go index 8dd8aded75..b32f6e46bd 100644 --- a/agents/drivers/cassandra-go/config_test.go +++ b/agents/drivers/cassandra-go/config_test.go @@ -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) } } } diff --git a/agents/drivers/cassandra-go/go.mod b/agents/drivers/cassandra-go/go.mod index 2da3d187af..f2929f5d59 100644 --- a/agents/drivers/cassandra-go/go.mod +++ b/agents/drivers/cassandra-go/go.mod @@ -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 +) diff --git a/agents/drivers/cassandra-go/go.sum b/agents/drivers/cassandra-go/go.sum index d5efef1665..511d4254cb 100644 --- a/agents/drivers/cassandra-go/go.sum +++ b/agents/drivers/cassandra-go/go.sum @@ -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= diff --git a/agents/drivers/cassandra-go/kerberos.go b/agents/drivers/cassandra-go/kerberos.go new file mode 100644 index 0000000000..c83c9b60d2 --- /dev/null +++ b/agents/drivers/cassandra-go/kerberos.go @@ -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 "" +} diff --git a/agents/drivers/cassandra-go/kerberos_test.go b/agents/drivers/cassandra-go/kerberos_test.go new file mode 100644 index 0000000000..8d57156a5c --- /dev/null +++ b/agents/drivers/cassandra-go/kerberos_test.go @@ -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 +}