Skip to content

Commit

Permalink
Changes the necessary data fields for the clientSSLCertSecret and ser…
Browse files Browse the repository at this point in the history 
…verSSLCertSecret (#850)
  • Loading branch information
@bartam1
bartam1 committed Aug 17, 2022
1 parent 98df9b3 commit 9b28b80
Show file tree
Hide file tree
Showing 11 changed files with 329 additions and 149 deletions.
7 changes: 2 additions & 5 deletions api/v1beta1/kafkacluster_types.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,9 +98,8 @@ type KafkaClusterSpec struct {
// It will be used by the koperator, cruise control, cruise control metrics reporter
// to communicate on SSL with that internal listener which is used for interbroker communication.
// The client certificate must share the same chain of trust as the server certificate used by the corresponding internal listener.
// The secret must contain the keystore, truststore jks files and the password for them in base64 encoded format and
// the tls certificate, tls private key, CA certificate in PEM format with base64 encoded
// under the keystore.jks, truststore.jks, password, tls.crt, tls.key, and ca.crt data fields.
// The secret must contain the keystore, truststore jks files and the password for them in base64 encoded format
// under the keystore.jks, truststore.jks, password data fields.
ClientSSLCertSecret *corev1.LocalObjectReference `json:"clientSSLCertSecret,omitempty"`
}

Expand Down Expand Up @@ -534,8 +533,6 @@ type CommonListenerSpec struct {
Type SecurityProtocol `json:"type"`
// ServerSSLCertSecret is a reference to the Kubernetes secret that contains the server certificate for the listener to be used for SSL communication.
// The secret must contain the keystore, truststore jks files and the password for them in base64 encoded format under the keystore.jks, truststore.jks, password data fields.
// When the listener is used for inner broker or controller communication the tls certificate is
// also needed in PEM format with base64 encoding under the tls.crt data field.
// If this field is omitted koperator will auto-create a self-signed server certificate using the configuration provided in 'sslSecrets' field.
ServerSSLCertSecret *corev1.LocalObjectReference `json:"serverSSLCertSecret,omitempty"`
// SSLClientAuth specifies whether client authentication is required, requested, or not required.
Expand Down
15 changes: 4 additions & 11 deletions charts/kafka-operator/templates/crds.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12273,9 +12273,8 @@ spec:
share the same chain of trust as the server certificate used by
the corresponding internal listener. The secret must contain the
keystore, truststore jks files and the password for them in base64
encoded format and the tls certificate, tls private key, CA certificate
in PEM format with base64 encoded under the keystore.jks, truststore.jks,
password, tls.crt, tls.key, and ca.crt data fields.
encoded format under the keystore.jks, truststore.jks, password
data fields.
properties:
name:
description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
Expand Down Expand Up @@ -19332,10 +19331,7 @@ spec:
to be used for SSL communication. The secret must contain
the keystore, truststore jks files and the password for
them in base64 encoded format under the keystore.jks,
truststore.jks, password data fields. When the listener
is used for inner broker or controller communication the
tls certificate is also needed in PEM format with base64
encoding under the tls.crt data field. If this field is
truststore.jks, password data fields. If this field is
omitted koperator will auto-create a self-signed server
certificate using the configuration provided in 'sslSecrets'
field.
Expand Down Expand Up @@ -19401,10 +19397,7 @@ spec:
to be used for SSL communication. The secret must contain
the keystore, truststore jks files and the password for
them in base64 encoded format under the keystore.jks,
truststore.jks, password data fields. When the listener
is used for inner broker or controller communication the
tls certificate is also needed in PEM format with base64
encoding under the tls.crt data field. If this field is
truststore.jks, password data fields. If this field is
omitted koperator will auto-create a self-signed server
certificate using the configuration provided in 'sslSecrets'
field.
Expand Down
15 changes: 4 additions & 11 deletions config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12272,9 +12272,8 @@ spec:
share the same chain of trust as the server certificate used by
the corresponding internal listener. The secret must contain the
keystore, truststore jks files and the password for them in base64
encoded format and the tls certificate, tls private key, CA certificate
in PEM format with base64 encoded under the keystore.jks, truststore.jks,
password, tls.crt, tls.key, and ca.crt data fields.
encoded format under the keystore.jks, truststore.jks, password
data fields.
properties:
name:
description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names
Expand Down Expand Up @@ -19331,10 +19330,7 @@ spec:
to be used for SSL communication. The secret must contain
the keystore, truststore jks files and the password for
them in base64 encoded format under the keystore.jks,
truststore.jks, password data fields. When the listener
is used for inner broker or controller communication the
tls certificate is also needed in PEM format with base64
encoding under the tls.crt data field. If this field is
truststore.jks, password data fields. If this field is
omitted koperator will auto-create a self-signed server
certificate using the configuration provided in 'sslSecrets'
field.
Expand Down Expand Up @@ -19400,10 +19396,7 @@ spec:
to be used for SSL communication. The secret must contain
the keystore, truststore jks files and the password for
them in base64 encoded format under the keystore.jks,
truststore.jks, password data fields. When the listener
is used for inner broker or controller communication the
tls certificate is also needed in PEM format with base64
encoding under the tls.crt data field. If this field is
truststore.jks, password data fields. If this field is
omitted koperator will auto-create a self-signed server
certificate using the configuration provided in 'sslSecrets'
field.
Expand Down
8 changes: 5 additions & 3 deletions pkg/pki/certmanagerpki/certmanager_tls_config_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,13 @@ func newMockControllerSecret(valid bool) *corev1.Secret {
secret.Name = "test-controller"
secret.Namespace = testNamespace
cert, key, _, _ := certutil.GenerateTestCert()
keystore_jks, password, _ := certutil.GenerateJKSFromByte(cert, key, cert)

if valid {
secret.Data = map[string][]byte{
corev1.TLSCertKey: cert,
corev1.TLSPrivateKeyKey: key,
v1alpha1.CoreCACertKey: cert,
v1alpha1.PasswordKey: password,
v1alpha1.TLSJKSTrustStore: keystore_jks,
v1alpha1.TLSJKSKeyStore: keystore_jks,
}
}
return secret
Expand Down
40 changes: 21 additions & 19 deletions pkg/resources/cruisecontrol/configmap.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ import (

const MinLogDirSizeInMB = int64(1)

func (r *Reconciler) configMap(clientPass, capacityConfig string, log logr.Logger) runtime.Object {
func (r *Reconciler) configMap(clientPass string, capacityConfig string, log logr.Logger) runtime.Object {
ccConfig := properties.NewProperties()

// Add base Cruise Control configuration
Expand Down Expand Up @@ -90,27 +90,29 @@ func (r *Reconciler) configMap(clientPass, capacityConfig string, log logr.Logge
return configMap
}

func generateSSLConfig(k v1beta1.KafkaClusterSpec, clientPass string, log logr.Logger) *properties.Properties {
sslConf := properties.NewProperties()

if k.IsClientSSLSecretPresent() && util.IsSSLEnabledForInternalCommunication(k.ListenersConfig.InternalListeners) {
if err := sslConf.Set("security.protocol", "SSL"); err != nil {
log.Error(err, "settings security.protocol in Cruise Control configuration failed")
}
if err := sslConf.Set("ssl.truststore.location", keystoreVolumePath+"/"+v1alpha1.TLSJKSTrustStore); err != nil {
log.Error(err, "settings ssl.truststore.location in Cruise Control configuration failed")
}
if err := sslConf.Set("ssl.keystore.location", keystoreVolumePath+"/"+v1alpha1.TLSJKSKeyStore); err != nil {
log.Error(err, "settings ssl.keystore.location in Cruise Control configuration failed")
func generateSSLConfig(kafkaCluster v1beta1.KafkaClusterSpec, clientPass string, log logr.Logger) *properties.Properties {
config := properties.NewProperties()
if kafkaCluster.IsClientSSLSecretPresent() && util.IsSSLEnabledForInternalCommunication(kafkaCluster.ListenersConfig.InternalListeners) {
keyStoreLoc := keystoreVolumePath + "/" + v1alpha1.TLSJKSKeyStore
trustStoreLoc := keystoreVolumePath + "/" + v1alpha1.TLSJKSTrustStore

sslConfig := map[string]string{
"security.protocol": "SSL",
"ssl.truststore.type": "JKS",
"ssl.keystore.type": "JKS",
"ssl.truststore.location": trustStoreLoc,
"ssl.keystore.location": keyStoreLoc,
"ssl.keystore.password": clientPass,
"ssl.truststore.password": clientPass,
}
if err := sslConf.Set("ssl.keystore.password", clientPass); err != nil {
log.Error(err, "settings ssl.keystore.password in Cruise Control configuration failed")
}
if err := sslConf.Set("ssl.truststore.password", clientPass); err != nil {
log.Error(err, "settings ssl.truststore.password in Cruise Control configuration failed")

for k, v := range sslConfig {
if err := config.Set(k, v); err != nil {
log.Error(err, fmt.Sprintf("setting %s parameter in cruise control configuration resulted an error", k))
}
}
}
return sslConf
return config
}

const (
Expand Down
58 changes: 39 additions & 19 deletions pkg/resources/cruisecontrol/cruisecontrol.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"github.com/banzaicloud/koperator/pkg/errorfactory"
"github.com/banzaicloud/koperator/pkg/k8sutil"
"github.com/banzaicloud/koperator/pkg/resources"
certutil "github.com/banzaicloud/koperator/pkg/util/cert"
pkicommon "github.com/banzaicloud/koperator/pkg/util/pki"
)

Expand Down Expand Up @@ -78,9 +79,14 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {

log.V(1).Info("Reconciling")

clientPass, err := r.getClientPassword()
if err != nil {
return err
var clientPass string
var err error

// Get configuration data from client secret
if r.KafkaCluster.Spec.IsClientSSLSecretPresent() {
if clientPass, err = r.getClientPassword(); err != nil {
return err
}
}

if r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint == "" {
Expand Down Expand Up @@ -148,25 +154,39 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
}

func (r *Reconciler) getClientPassword() (string, error) {
var clientPass string
if r.KafkaCluster.Spec.IsClientSSLSecretPresent() {
// Use that secret as default which has autogenerated for clients by us
clientNamespacedName := types.NamespacedName{Name: fmt.Sprintf(pkicommon.BrokerControllerTemplate, r.KafkaCluster.Name), Namespace: r.KafkaCluster.Namespace}
if r.KafkaCluster.Spec.GetClientSSLCertSecretName() != "" {
clientNamespacedName = types.NamespacedName{Name: r.KafkaCluster.Spec.GetClientSSLCertSecretName(), Namespace: r.KafkaCluster.Namespace}
clientSecret, err := r.getClientSecret()
if err != nil {
return "", err
}
return string(clientSecret.Data[v1alpha1.PasswordKey]), nil
}

func (r *Reconciler) getClientSecret() (*corev1.Secret, error) {
clientSecret := &corev1.Secret{}
// Use that secret as default which has autogenerated for clients by us
clientNamespacedName := types.NamespacedName{Name: fmt.Sprintf(pkicommon.BrokerControllerTemplate, r.KafkaCluster.Name), Namespace: r.KafkaCluster.Namespace}
if r.KafkaCluster.Spec.GetClientSSLCertSecretName() != "" {
clientNamespacedName = types.NamespacedName{Name: r.KafkaCluster.Spec.GetClientSSLCertSecretName(), Namespace: r.KafkaCluster.Namespace}
}

if err := r.Client.Get(context.Background(), clientNamespacedName, clientSecret); err != nil {
// We only return with ResourceNotReady (which is going to retry after a period time)
// when we use our cert generation for client cert
if apierrors.IsNotFound(err) && r.KafkaCluster.Spec.GetClientSSLCertSecretName() == "" {
return nil, errorfactory.New(errorfactory.ResourceNotReady{}, err, "client secret not ready")
}
clientSecret := &corev1.Secret{}
if err := r.Client.Get(context.TODO(), clientNamespacedName, clientSecret); err != nil {
// We only return with ResourceNotReady (which is goin to retry after period time)
// when we use our cert generation for client cert
if apierrors.IsNotFound(err) && r.KafkaCluster.Spec.GetClientSSLCertSecretName() == "" {
return "", errorfactory.New(errorfactory.ResourceNotReady{}, err, "client secret not ready")
}
return "", errors.WrapIfWithDetails(err, "failed to get client secret")

return nil, errors.WrapIfWithDetails(err, "failed to get client secret")
}

if err := certutil.CheckSSLCertSecret(clientSecret); err != nil {
if r.KafkaCluster.Spec.GetClientSSLCertSecretName() == "" {
return nil, errorfactory.New(errorfactory.ResourceNotReady{}, errors.Errorf("SSL JKS certificate has not generated properly yet into client secret: %s", clientSecret.Name), "checking secret data fields")
}
clientPass = string(clientSecret.Data[v1alpha1.PasswordKey])
return nil, errors.WrapIfWithDetails(err, "failed to get certificates from client secret")
}
return clientPass, nil

return clientSecret, nil
}

func isBrokerDeletionInProgress(brokerState map[string]v1beta1.BrokerState) bool {
Expand Down

0 comments on commit 9b28b80

Please sign in to comment.