Develop

.gitignore

@@ -27,3 +27,4 @@ openspec/
 .opencode/
 AGENTS.md
 CLAUDE.md
+opencode.json

README.md

@@ -84,11 +84,14 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
 
 **TLS Configuration Options:**
 
-- `tls.enabled` - Enable/disable TLS for Kafka connections (default: false)
-- `tls.cert_file` - Path to client certificate file for mTLS authentication (optional)
-- `tls.key_file` - Path to client private key file for mTLS authentication (optional)
-- `tls.ca_cert_file` - Path to CA certificate file for broker verification (optional)
-- `tls.insecure_skip_verify` - Skip certificate verification (insecure, for testing only, default: false)
+- `tls_enabled` - Enable/disable TLS for Kafka connections (default: false)
+- `tls_cert_file` - Path to client certificate file for mTLS authentication (optional)
+- `tls_key_file` - Path to client private key file for mTLS authentication (optional)
+- `tls_ca_cert_file` - Path to CA certificate file for broker verification (optional)
+- `tls_insecure_skip_verify` - Skip certificate verification (insecure, for testing only, default: false)
+- `tls_server_name` - Optional SNI and certificate name override (default: empty)
+
+Kafka keeps its existing `tls_enabled` flag for each consumer cluster and producer. Cassandra uses its existing `is_ssl_enabled` flag; after TLS is enabled, both integrations use the same flat `tls_*` fields, including identical `tls_insecure_skip_verify` and `tls_server_name` behavior.
 
 **Consumer TLS Configuration Example:**
 
@@ -99,13 +102,12 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
         topics = "config-version-report"
         consumer_group = "webconfig"
 
-        tls {
-            enabled = true
-            cert_file = "/etc/webconfig/kafka/client.crt"
-            key_file = "/etc/webconfig/kafka/client.key"
-            ca_cert_file = "/etc/webconfig/kafka/ca.crt"
-            insecure_skip_verify = false
-        }
+        tls_enabled = true
+        tls_cert_file = "/etc/webconfig/kafka/client.crt"
+        tls_key_file = "/etc/webconfig/kafka/client.key"
+        tls_ca_cert_file = "/etc/webconfig/kafka/ca.crt"
+        tls_insecure_skip_verify = false
+        tls_server_name = "kafka.example.com"
 
         # Per-cluster TLS configuration
         clusters {
@@ -114,12 +116,12 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
                 brokers = "kafka-mesh:9093"
                 topics = "staging-chi-onewifi-from-device"
 
-                tls {
-                    enabled = true
-                    cert_file = "/etc/webconfig/kafka/mesh-client.crt"
-                    key_file = "/etc/webconfig/kafka/mesh-client.key"
-                    ca_cert_file = "/etc/webconfig/kafka/mesh-ca.crt"
-                }
+                tls_enabled = true
+                tls_cert_file = "/etc/webconfig/kafka/mesh-client.crt"
+                tls_key_file = "/etc/webconfig/kafka/mesh-client.key"
+                tls_ca_cert_file = "/etc/webconfig/kafka/mesh-ca.crt"
+                tls_insecure_skip_verify = false
+                tls_server_name = "kafka-mesh.example.com"
             }
         }
     }
@@ -130,23 +132,27 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
 ```shell
     kafka_producer {
         enabled = true
+        // one_line consolidates consumer and producer enqueue success under logger=kafka;
+        // two_line also emits the consumer processing success record and uses
+        // logger=kafkaproducer for producer records.
+        logging_mode = "one_line"
         brokers = "kafka-broker:9093"
         topic = "webconfig_downstream"
 
-        tls {
-            enabled = true
-            cert_file = "/etc/webconfig/kafka/producer-client.crt"
-            key_file = "/etc/webconfig/kafka/producer-client.key"
-            ca_cert_file = "/etc/webconfig/kafka/ca.crt"
-        }
+        tls_enabled = true
+        tls_cert_file = "/etc/webconfig/kafka/producer-client.crt"
+        tls_key_file = "/etc/webconfig/kafka/producer-client.key"
+        tls_ca_cert_file = "/etc/webconfig/kafka/ca.crt"
+        tls_insecure_skip_verify = false
+        tls_server_name = "kafka.example.com"
     }
 ```
 
 **Certificate Requirements:**
 
-1. **Client Certificate (mTLS)**: If `cert_file` and `key_file` are provided, mutual TLS authentication is enabled. The certificate and key must be in PEM format.
+1. **Client Certificate (mTLS)**: If `tls_cert_file` and `tls_key_file` are provided, mutual TLS authentication is enabled. The certificate and key must be in PEM format.
 
-2. **CA Certificate**: If `ca_cert_file` is provided, it will be used to verify the Kafka broker's certificate. This is useful when using self-signed certificates or internal CAs.
+2. **CA Certificate**: If `tls_ca_cert_file` is provided, it will be used to verify the Kafka broker's certificate. This is useful when using self-signed certificates or internal CAs.
 
 3. **Certificate Validation**: All certificate files are validated at startup. The application will fail to start with clear error messages if:
    - Certificate files are missing or unreadable
@@ -156,8 +162,8 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
 **TLS Security Best Practices:**
 
 1. **Use TLS in Production**: Always enable TLS for production Kafka connections to encrypt data in transit
-2. **Use mTLS**: Provide client certificates (`cert_file` and `key_file`) for mutual authentication
-3. **Verify Certificates**: Never use `insecure_skip_verify = true` in production - it disables certificate verification and is insecure
+2. **Use mTLS**: Provide client certificates (`tls_cert_file` and `tls_key_file`) for mutual authentication
+3. **Verify Certificates**: Never use `tls_insecure_skip_verify = true` in production - it disables certificate verification and is insecure
 4. **Protect Certificate Files**: Set appropriate file permissions (0600) on certificate and key files
 5. **Use Secure Ports**: Configure Kafka brokers to listen on secure ports (typically 9093 for TLS)
 6. **Certificate Rotation**: Plan for certificate rotation - the service must be restarted to pick up new certificates

common/kafka_tls.go

@@ -27,109 +27,115 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-// LoadKafkaTLSConfig loads TLS configuration from HOCON config at the specified prefix.
-// Prefix should be like "webconfig.kafka" or "webconfig.kafka.clusters.mesh" or "webconfig.kafka_producer".
-// Returns nil if TLS is not enabled.
-// Returns error if TLS is enabled but configuration is invalid.
-func LoadKafkaTLSConfig(conf *configuration.Config, prefix string) (*tls.Config, error) {
-	tlsEnabled := conf.GetBoolean(prefix + ".tls_enabled")
-	if !tlsEnabled {
-		return nil, nil
+// LoadTLSConfig loads the shared flat tls_* configuration used by Kafka and Cassandra.
+// Prefix should be empty for a driver sub-config, or a full HOCON path such as
+// "webconfig.kafka" or "webconfig.kafka_producer".
+func LoadTLSConfig(conf *configuration.Config, prefix, component string) (*tls.Config, error) {
+	key := func(name string) string {
+		if prefix == "" {
+			return name
+		}
+		return prefix + "." + name
 	}
 
-	tlsConfig := &tls.Config{}
-
-	// Check insecure_skip_verify flag first
-	insecureSkipVerify := conf.GetBoolean(prefix + ".tls_insecure_skip_verify")
+	insecureSkipVerify := conf.GetBoolean(key("tls_insecure_skip_verify"))
+	certFile := conf.GetString(key("tls_cert_file"))
+	keyFile := conf.GetString(key("tls_key_file"))
+	caCertFile := conf.GetString(key("tls_ca_cert_file"))
+	serverName := conf.GetString(key("tls_server_name"))
+
+	cipherSuites := []uint16{
+		tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
+		tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
+	}
+	if component == "Cassandra" {
+		cipherSuites = append(cipherSuites, tls.TLS_RSA_WITH_AES_128_CBC_SHA)
+	}
 
-	// Load client certificates for mTLS if provided (optional when insecure_skip_verify is true)
-	certFile := conf.GetString(prefix + ".tls_cert_file")
-	keyFile := conf.GetString(prefix + ".tls_key_file")
+	tlsConfig := &tls.Config{
+		MinVersion:         tls.VersionTLS12,
+		ServerName:         serverName,
+		CipherSuites:       cipherSuites,
+		InsecureSkipVerify: insecureSkipVerify,
+	}
 
-	// When insecure_skip_verify is true and no cert files configured, skip loading certificates
-	// This allows TLS without client authentication (server-only TLS)
 	if insecureSkipVerify && (len(certFile) == 0 || len(keyFile) == 0) {
-		// Insecure mode without client certificates - skip cert loading
+		log.WithField("component", component).Warn("TLS enabled in insecure mode without client certificates")
 	} else if len(certFile) > 0 && len(keyFile) > 0 {
-		// Only validate cert files exist when verification is enabled
 		if !insecureSkipVerify {
-			// Validate certificate file exists
 			if _, err := os.Stat(certFile); os.IsNotExist(err) {
 				return nil, NewError(fmt.Errorf("TLS certificate file does not exist: %s", certFile))
 			}
-
-			// Validate key file exists
 			if _, err := os.Stat(keyFile); os.IsNotExist(err) {
 				return nil, NewError(fmt.Errorf("TLS key file does not exist: %s", keyFile))
 			}
 		}
 
-		// Load and parse the certificate and key
 		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
 		if err != nil {
 			return nil, NewError(fmt.Errorf("failed to load TLS certificate and key from %s and %s: %v", certFile, keyFile, err))
 		}
-
 		tlsConfig.Certificates = []tls.Certificate{cert}
 		log.WithFields(log.Fields{
-			"prefix":    prefix,
+			"component": component,
 			"cert_file": certFile,
 			"key_file":  keyFile,
 		}).Info("Loaded TLS client certificate for mTLS")
 	} else if len(certFile) > 0 || len(keyFile) > 0 {
-		// Partial cert configuration detected - require both cert and key when verification is enabled
 		if !insecureSkipVerify {
 			return nil, NewError(fmt.Errorf("TLS enabled with verification but incomplete certificate configuration (cert: %s, key: %s)", certFile, keyFile))
 		}
 	}
 
-	// Load CA certificate if provided (optional when insecure_skip_verify is true)
-	caCertFile := conf.GetString(prefix + ".tls_ca_cert_file")
-	// When insecure_skip_verify is true and no CA file configured, skip loading CA cert
-	// This allows TLS without broker verification (insecure mode)
 	if insecureSkipVerify && len(caCertFile) == 0 {
-		// Insecure mode without CA cert - skip CA loading
+		log.WithField("component", component).Warn("TLS enabled in insecure mode without CA certificate")
 	} else if len(caCertFile) > 0 {
-		// Only validate CA cert file exists when verification is enabled
 		if !insecureSkipVerify {
-			// Validate CA certificate file exists
 			if _, err := os.Stat(caCertFile); os.IsNotExist(err) {
 				return nil, NewError(fmt.Errorf("TLS CA certificate file does not exist: %s", caCertFile))
 			}
 		}
 
-		// Load CA certificate
 		caCert, err := os.ReadFile(caCertFile)
 		if err != nil {
 			return nil, NewError(fmt.Errorf("failed to read TLS CA certificate from %s: %v", caCertFile, err))
 		}
-
-		// Parse CA certificate
 		caCertPool := x509.NewCertPool()
 		if !caCertPool.AppendCertsFromPEM(caCert) {
 			return nil, NewError(fmt.Errorf("failed to parse TLS CA certificate from %s", caCertFile))
 		}
-
 		tlsConfig.RootCAs = caCertPool
 		log.WithFields(log.Fields{
-			"prefix":       prefix,
+			"component":    component,
 			"ca_cert_file": caCertFile,
-		}).Info("Loaded TLS CA certificate for broker verification")
+		}).Info("Loaded TLS CA certificate for server verification")
 	}
 
 	if insecureSkipVerify {
-		tlsConfig.InsecureSkipVerify = true
-		log.WithFields(log.Fields{
-			"prefix": prefix,
-		}).Warn("TLS certificate verification is disabled (insecure_skip_verify=true). This is insecure and should only be used for testing.")
+		log.WithField("component", component).Warn("TLS certificate verification is disabled (insecure_skip_verify=true). This is insecure and should only be used for testing.")
 	}
-
 	log.WithFields(log.Fields{
-		"prefix":               prefix,
+		"component":            component,
 		"has_client_cert":      len(tlsConfig.Certificates) > 0,
 		"has_ca_cert":          tlsConfig.RootCAs != nil,
 		"insecure_skip_verify": insecureSkipVerify,
-	}).Info("TLS configuration loaded for Kafka connection")
+		"server_name":          serverName,
+	}).Info("TLS configuration loaded")
 
 	return tlsConfig, nil
 }
+
+// LoadKafkaTLSConfig loads enabled Kafka TLS configuration from HOCON.
+// Prefix should be like "webconfig.kafka", "webconfig.kafka.clusters.mesh",
+// or "webconfig.kafka_producer".
+func LoadKafkaTLSConfig(conf *configuration.Config, prefix string) (*tls.Config, error) {
+	tlsEnabled := conf.GetBoolean(prefix + ".tls_enabled")
+	if !tlsEnabled {
+		return nil, nil
+	}
+	return LoadTLSConfig(conf, prefix, "Kafka")
+}

common/kafka_tls_test.go

@@ -20,6 +20,7 @@ package common
 import (
 	"crypto/rand"
 	"crypto/rsa"
+	"crypto/tls"
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/pem"
@@ -325,6 +326,7 @@ webconfig {
 		tls_cert_file = "` + certFile + `"
 		tls_key_file = "` + keyFile + `"
 		tls_ca_cert_file = "` + caCertFile + `"
+			tls_server_name = "kafka.example.com"
 	}
 }
 `
@@ -336,6 +338,38 @@ webconfig {
 	assert.Assert(t, len(tlsConfig.Certificates) == 1, "Should have one client certificate")
 	assert.Assert(t, tlsConfig.RootCAs != nil, "Should have custom CA")
 	assert.Assert(t, !tlsConfig.InsecureSkipVerify, "Should not skip verification")
+	assert.Equal(t, tlsConfig.ServerName, "kafka.example.com")
+	assert.Equal(t, tlsConfig.MinVersion, uint16(0x0303))
+	assert.Assert(t, len(tlsConfig.CipherSuites) > 0, "Should configure TLS cipher suites")
+	assert.Assert(t, !containsCipherSuite(tlsConfig.CipherSuites, tls.TLS_RSA_WITH_AES_128_CBC_SHA), "Kafka should not enable the Cassandra legacy cipher suite")
+}
+
+func TestLoadTLSConfig_CassandraIncludesLegacyCipher(t *testing.T) {
+	conf := configuration.ParseString(`
+webconfig {
+	database {
+		cassandra {
+			tls_server_name = "cassandra.example.com"
+			tls_insecure_skip_verify = false
+		}
+	}
+}
+`)
+
+	tlsConfig, err := LoadTLSConfig(conf.GetConfig("webconfig.database.cassandra"), "", "Cassandra")
+
+	assert.NilError(t, err)
+	assert.Equal(t, tlsConfig.ServerName, "cassandra.example.com")
+	assert.Assert(t, containsCipherSuite(tlsConfig.CipherSuites, tls.TLS_RSA_WITH_AES_128_CBC_SHA), "Cassandra should retain its legacy 3.11.x cipher suite")
+}
+
+func containsCipherSuite(cipherSuites []uint16, wanted uint16) bool {
+	for _, cipherSuite := range cipherSuites {
+		if cipherSuite == wanted {
+			return true
+		}
+	}
+	return false
 }
 
 func TestLoadKafkaTLSConfig_DifferentPrefixes(t *testing.T) {

config/sample_webconfig.conf

@@ -162,6 +162,12 @@ webconfig {
             enabled = false
         }
 
+        // Protect /api/v1/token with API token authentication.
+        // Defaults to false to preserve existing endpoint access.
+        token_api_token_auth {
+            enabled = false
+        }
+
         device_api_token_auth {
             enabled = true
         }
@@ -286,14 +292,16 @@ webconfig {
 
         // TLS configuration for secure Kafka connections
         // NOTE: These are flat keys read by LoadKafkaTLSConfig at prefix "webconfig.kafka"
-        // Certificate files (tls_cert_file, tls_key_file, tls_ca_cert_file) are REQUIRED when
-        // tls_insecure_skip_verify=false (default). They are OPTIONAL when tls_insecure_skip_verify=true.
+        // Client certificate/key are optional unless the Kafka broker requires mTLS.
+        // tls_ca_cert_file is optional; when unset, the system CA pool is used.
         tls_enabled = false
         tls_cert_file = "/etc/webconfig/kafka/client-cert.pem"
         tls_key_file = "/etc/webconfig/kafka/client-key.pem"
         tls_ca_cert_file = "/etc/webconfig/kafka/ca-cert.pem"
         // WARNING: set true only in dev/test environments, NEVER in production
         tls_insecure_skip_verify = false
+        // Optional SNI/server certificate name. Leave empty to use the broker address.
+        tls_server_name = ""
 
         // Reconnect resilience — tuned for broker restart recovery.
 
@@ -359,6 +367,7 @@ webconfig {
                 tls_key_file = "/etc/webconfig/kafka/mesh-client-key.pem"
                 tls_ca_cert_file = "/etc/webconfig/kafka/mesh-ca-cert.pem"
                 tls_insecure_skip_verify = false
+                tls_server_name = ""
             }
             east {
                 enabled = false
@@ -379,6 +388,7 @@ webconfig {
                 tls_key_file = "/etc/webconfig/kafka/east-client-key.pem"
                 tls_ca_cert_file = "/etc/webconfig/kafka/east-ca-cert.pem"
                 tls_insecure_skip_verify = false
+                tls_server_name = ""
             }
         }
     }
@@ -406,6 +416,10 @@ webconfig {
     // forward kafka messages if needed
     kafka_producer {
         enabled = false
+        // Kafka success logging: "one_line" consolidates consumer and producer enqueue success
+        // under logger=kafka; "two_line" also emits the consumer processing success record and
+        // uses logger=kafkaproducer for producer records.
+        logging_mode = "one_line"
         brokers = "localhost:9092"
         topic = "webconfig_downstream"
 
@@ -415,6 +429,7 @@ webconfig {
         tls_key_file = "/etc/webconfig/kafka/producer-client-key.pem"
         tls_ca_cert_file = "/etc/webconfig/kafka/producer-ca-cert.pem"
         tls_insecure_skip_verify = false
+        tls_server_name = ""
 
         // Reconnect resilience — tuned for broker restart recovery.