Kafka auth support

cert.pem

@@ -0,0 +1,24 @@
+-----BEGIN CERTIFICATE-----
+MIID/jCCAuagAwIBAgIUTsdR5JCRLZx8gNn5/Hc4Lle9pB8wDQYJKoZIhvcNAQEL
+BQAwdDELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMRYwFAYDVQQHDA1TYW4gRnJh
+bmNpc2NvMREwDwYDVQQKDAhLZXljbG9hazEtMCsGA1UEAwwkc2ltcGxlLWtleWNs
+b2FrLmFwcHMuY2x1c3Rlci5leGFtcGxlMB4XDTI1MDEzMTA3MjI1MVoXDTI2MDEz
+MTA3MjI1MVowdDELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMRYwFAYDVQQHDA1T
+YW4gRnJhbmNpc2NvMREwDwYDVQQKDAhLZXljbG9hazEtMCsGA1UEAwwkc2ltcGxl
+LWtleWNsb2FrLmFwcHMuY2x1c3Rlci5leGFtcGxlMIIBIjANBgkqhkiG9w0BAQEF
+AAOCAQ8AMIIBCgKCAQEAq1Kouy7RTmfigX8pfgItDoqAZtBxQmXl8yqLsfNIabcy
+mFWDptAeZQErGMOjDozSPbfPacTBRr/CBtOhNpGxojMN7F9gH0aZDQfNmxD5lBJb
+RIKD6pcvSeS5fkP+GKtp0wtfjTG27kA41+K0uYuVOAHDDweEOjQIiuUX40yZilkx
+xk9Tatz0BgptUW6WLY1MYYfju0wHdfMf1oGbDBuX8lCGF+vHWpyEARjfYBrxKGYa
+UW+9Isui2YnOXIDyCDZngP1ctmt+jc7JZCPZdAkPS2I8w/WLLmQetryvnSdd+Run
+vIPhaEvaWV018tvOA24j8sRTm1T7NPjvFQoDJacb9QIDAQABo4GHMIGEMGMGA1Ud
+EQRcMFqCJHNpbXBsZS1rZXljbG9hay5hcHBzLmNsdXN0ZXIuZXhhbXBsZYIIa2V5
+Y2xvYWuCDGtleWNsb2FrLnN2Y4Iaa2V5Y2xvYWsuc3ZjLmNsdXN0ZXIubG9jYWww
+HQYDVR0OBBYEFKnx5Hc6DfUqIaXc2fvq6FbVG6WPMA0GCSqGSIb3DQEBCwUAA4IB
+AQAMEcYAgL7T1IQdG12pnvwDP1rcNeIGdPgZlVlmXvvM57MGHVP06D2SkbyAiQlQ
+r6Ohze4oS+9VcrkqAqoo2H3/fPT/dFLywqrXW47bs/7aEtD7bGnfwCkgS2mhn4ZK
+9vpWRrrKQ4ZAG8YG4rJFbsaMNAhMB/joUjjLh60RoQnDdWmyhAbl07ARzCQwuv8M
+DvpzrMi/U8nBFlvUPReIpsejj9HAyyfRBv2dJwSZGLxFOb2flSIxSC8aaaJ8NQmD
+TZeZInxtUvHwEnnH0u5pB3wE+u8NklGSu7Xo3ctK+9q4Cb51vGapwd6dhtjnmlmx
+gg82StaL1n9k01Rhu9keXA3J
+-----END CERTIFICATE-----

operator/controller/src/main/java/io/apicurio/registry/operator/Constants.java

@@ -36,6 +36,16 @@ public class Constants {
             .withInitialDelaySeconds(15).withTimeoutSeconds(5).withPeriodSeconds(10).withSuccessThreshold(1)
             .withFailureThreshold(3).build();
 
+    public static final Probe TLS_DEFAULT_READINESS_PROBE = new ProbeBuilder().withNewHttpGet()
+            .withScheme("HTTPS").withPath("/health/ready").withNewPort().withValue(8443).endPort().endHttpGet()
+            .withInitialDelaySeconds(15).withTimeoutSeconds(5).withPeriodSeconds(10).withSuccessThreshold(1)
+            .withFailureThreshold(3).build();
+
+    public static final Probe TLS_DEFAULT_LIVENESS_PROBE = new ProbeBuilder().withNewHttpGet()
+            .withScheme("HTTPS").withPath("/health/live").withNewPort().withValue(8443).endPort().endHttpGet()
+            .withInitialDelaySeconds(15).withTimeoutSeconds(5).withPeriodSeconds(10).withSuccessThreshold(1)
+            .withFailureThreshold(3).build();
+
     public static final Map<String, String> BASIC_LABELS = Map.of(MANAGED_BY_LABEL, MANAGED_BY_VALUE,
             LABEL_SELECTOR_KEY, LABEL_SELECTOR_VALUE);
 

operator/controller/src/main/java/io/apicurio/registry/operator/EnvironmentVariables.java

@@ -6,10 +6,16 @@ public class EnvironmentVariables {
     public static final String QUARKUS_HTTP_ACCESS_LOG_ENABLED = "QUARKUS_HTTP_ACCESS_LOG_ENABLED";
     public static final String QUARKUS_HTTP_CORS_ORIGINS = "QUARKUS_HTTP_CORS_ORIGINS";
 
+    public static final String  QUARKUS_HTTP_INSECURE_REQUESTS = "QUARKUS_HTTP_INSECURE_REQUESTS";
+    public static final String QUARKUS_TLS_KEY_STORE_P12_PATH = "QUARKUS_TLS_KEY_STORE_P12_PATH";
+    public static final String QUARKUS_TLS_KEY_STORE_P12_PASSWORD = "QUARKUS_TLS_KEY_STORE_P12_PASSWORD";
+    public static final String QUARKUS_TLS_TRUST_STORE_P12_PATH = "QUARKUS_TLS_TRUST_STORE_P12_PATH";
+    public static final String QUARKUS_TLS_TRUST_STORE_P12_PASSWORD = "QUARKUS_TLS_TRUST_STORE_P12_PASSWORD";
+    public static final String QUARKUS_OIDC_TLS_TLS_CONFIGURATION_NAME = "QUARKUS_OIDC_TLS_TLS_CONFIGURATION_NAME";
+
     public static final String APICURIO_REST_DELETION_ARTIFACT_VERSION_ENABLED = "APICURIO_REST_DELETION_ARTIFACT-VERSION_ENABLED";
     public static final String APICURIO_REST_DELETION_ARTIFACT_ENABLED = "APICURIO_REST_DELETION_ARTIFACT_ENABLED";
     public static final String APICURIO_REST_DELETION_GROUP_ENABLED = "APICURIO_REST_DELETION_GROUP_ENABLED";
-
     public static final String APICURIO_REST_MUTABILITY_ARTIFACT_VERSION_CONTENT_ENABLED = "APICURIO_REST_MUTABILITY_ARTIFACT-VERSION-CONTENT_ENABLED";
 
     private static final String KAFKA_PREFIX = "APICURIO_KAFKA_COMMON_";
@@ -21,6 +27,14 @@ public class EnvironmentVariables {
     public static final String KAFKASQL_SSL_TRUSTSTORE_LOCATION = KAFKA_PREFIX + "SSL_TRUSTSTORE_LOCATION";
     public static final String KAFKASQL_SSL_TRUSTSTORE_PASSWORD = KAFKA_PREFIX + "SSL_TRUSTSTORE_PASSWORD";
 
+    // KafkaSQL oauth
+    public static final String APICURIO_KAFKASQL_SECURITY_SASL_ENABLED = "APICURIO_KAFKASQL_SECURITY_SASL_ENABLED";
+    public static final String APICURIO_KAFKASQL_SECURITY_SASL_MECHANISM = "APICURIO_KAFKASQL_SECURITY_SASL_MECHANISM";
+    public static final String APICURIO_KAFKASQL_SECURITY_SASL_CLIENT_ID = "APICURIO_KAFKASQL_SECURITY_SASL_CLIENT_ID";
+    public static final String APICURIO_KAFKASQL_SECURITY_SASL_CLIENT_SECRET = "APICURIO_KAFKASQL_SECURITY_SASL_CLIENT_SECRET";
+    public static final String APICURIO_KAFKASQL_SECURITY_SASL_TOKEN_ENDPOINT = "APICURIO_KAFKASQL_SECURITY_SASL_TOKEN_ENDPOINT";
+    public static final String APICURIO_KAFKASQL_SECURITY_SASL_LOGIN_CALLBACK_HANDLER_CLASS = "APICURIO_KAFKASQL_SECURITY_SASL_LOGIN_CALLBACK_HANDLER_CLASS";
+
     // Auth related environment variables
     public static final String APICURIO_REGISTRY_AUTH_ENABLED = "QUARKUS_OIDC_TENANT_ENABLED";
     public static final String APICURIO_REGISTRY_APP_CLIENT_ID = "QUARKUS_OIDC_CLIENT_ID";

operator/controller/src/main/java/io/apicurio/registry/operator/feat/KafkaSql.java

@@ -12,6 +12,7 @@
 
 import java.util.Map;
 
+import static io.apicurio.registry.operator.EnvironmentVariables.KAFKASQL_SECURITY_PROTOCOL;
 import static io.apicurio.registry.operator.api.v1.ContainerNames.REGISTRY_APP_CONTAINER_NAME;
 import static io.apicurio.registry.operator.resource.app.AppDeploymentResource.addEnvVar;
 import static io.apicurio.registry.operator.utils.Utils.isBlank;
@@ -25,7 +26,7 @@ public class KafkaSql {
     public static String ENV_KAFKASQL_BOOTSTRAP_SERVERS = "APICURIO_KAFKASQL_BOOTSTRAP_SERVERS";
 
     public static void configureKafkaSQL(ApicurioRegistry3 primary, Deployment deployment,
-            Map<String, EnvVar> env) {
+                                         Map<String, EnvVar> env) {
         ofNullable(primary.getSpec()).map(ApicurioRegistry3Spec::getApp).map(AppSpec::getStorage)
                 .map(StorageSpec::getKafkasql).ifPresent(kafkasql -> {
                     if (!isBlank(kafkasql.getBootstrapServers())) {
@@ -34,10 +35,30 @@ public static void configureKafkaSQL(ApicurioRegistry3 primary, Deployment deplo
                         addEnvVar(env, new EnvVarBuilder().withName(ENV_KAFKASQL_BOOTSTRAP_SERVERS)
                                 .withValue(kafkasql.getBootstrapServers()).build());
 
-                        if (KafkaSqlTLS.configureKafkaSQLTLS(primary, deployment, REGISTRY_APP_CONTAINER_NAME,
-                                env)) {
+                        boolean sslConfigured = KafkaSqlTLS.configureKafkaSQLTLS(primary, deployment, REGISTRY_APP_CONTAINER_NAME,
+                                env);
+
+                        boolean oAuthConfigured = KafkaSqlAuth.configureKafkaSQLOauth(primary,
+                                env);
+
+                        if (sslConfigured) {
                             log.info("KafkaSQL storage with TLS security configured.");
                         }
+
+                        if (oAuthConfigured) {
+                            log.info("KafkaSQL storage with Oauth security configured.");
+                        }
+
+                        // Set the security protocol
+                         if (sslConfigured) {
+                            if (oAuthConfigured) {
+                                addEnvVar(env, KAFKASQL_SECURITY_PROTOCOL, "SASL_SSL");
+                            } else {
+                                addEnvVar(env, KAFKASQL_SECURITY_PROTOCOL, "SSL");
+                            }
+                        } else if (oAuthConfigured) {
+                            addEnvVar(env, KAFKASQL_SECURITY_PROTOCOL, "SASL_PLAINTEXT");
+                        }
                     }
                 });
     }

operator/controller/src/main/java/io/apicurio/registry/operator/feat/KafkaSqlAuth.java

@@ -0,0 +1,64 @@
+package io.apicurio.registry.operator.feat;
+
+import io.apicurio.registry.operator.api.v1.ApicurioRegistry3;
+import io.apicurio.registry.operator.api.v1.ApicurioRegistry3Spec;
+import io.apicurio.registry.operator.api.v1.spec.AppSpec;
+import io.apicurio.registry.operator.api.v1.spec.KafkaSqlAuthSpec;
+import io.apicurio.registry.operator.api.v1.spec.KafkaSqlSpec;
+import io.apicurio.registry.operator.api.v1.spec.StorageSpec;
+import io.apicurio.registry.operator.utils.SecretKeyRefTool;
+import io.fabric8.kubernetes.api.model.EnvVar;
+
+import java.util.Map;
+import java.util.Optional;
+
+import static io.apicurio.registry.operator.EnvironmentVariables.*;
+import static io.apicurio.registry.operator.resource.app.AppDeploymentResource.addEnvVar;
+import static java.util.Optional.ofNullable;
+
+public class KafkaSqlAuth {
+
+    /**
+     * KafkaSQL must be already configured.
+     */
+    public static boolean configureKafkaSQLOauth(ApicurioRegistry3 primary, Map<String, EnvVar> env) {
+
+        // spotless:off
+        var clientSecret = new SecretKeyRefTool(getKafkaSqlAuthSpec(primary)
+                .map(KafkaSqlAuthSpec::getClientSecretRef)
+                .orElse(null), "clientSecret");
+
+        var clientId = new SecretKeyRefTool(getKafkaSqlAuthSpec(primary)
+                .map(KafkaSqlAuthSpec::getClientIdRef)
+                .orElse(null), "clientId");
+
+        if (clientSecret.isValid()) {
+            getKafkaSqlAuthSpec(primary)
+                    .filter(KafkaSqlAuthSpec::getEnabled)
+                    .ifPresent(kafkaSqlAuthSpec -> {
+                        addEnvVar(env, APICURIO_KAFKASQL_SECURITY_SASL_ENABLED, kafkaSqlAuthSpec.getEnabled().toString());
+                        addEnvVar(env, APICURIO_KAFKASQL_SECURITY_SASL_MECHANISM, kafkaSqlAuthSpec.getMechanism());
+
+                        clientId.applySecretEnvVar(env, APICURIO_KAFKASQL_SECURITY_SASL_CLIENT_ID);
+                        clientSecret.applySecretEnvVar(env, APICURIO_KAFKASQL_SECURITY_SASL_CLIENT_SECRET);
+
+                        addEnvVar(env, APICURIO_KAFKASQL_SECURITY_SASL_TOKEN_ENDPOINT, kafkaSqlAuthSpec.getTokenEndpoint());
+                        addEnvVar(env, APICURIO_KAFKASQL_SECURITY_SASL_LOGIN_CALLBACK_HANDLER_CLASS, kafkaSqlAuthSpec.getLoginHandlerClass());
+                    });
+
+            return true;
+        }
+        return false;
+    }
+
+    private static Optional<KafkaSqlAuthSpec> getKafkaSqlAuthSpec(ApicurioRegistry3 primary) {
+        // spotless:off
+        return ofNullable(primary)
+                .map(ApicurioRegistry3::getSpec)
+                .map(ApicurioRegistry3Spec::getApp)
+                .map(AppSpec::getStorage)
+                .map(StorageSpec::getKafkasql)
+                .map(KafkaSqlSpec::getAuth);
+        // spotless:on
+    }
+}

operator/controller/src/main/java/io/apicurio/registry/operator/feat/KafkaSqlTLS.java

@@ -41,28 +41,32 @@ public static boolean configureKafkaSQLTLS(ApicurioRegistry3 primary, Deployment
                 .map(KafkaSqlTLSSpec::getTruststorePasswordSecretRef)
                 .orElse(null), "ca.password");
 
-        if (truststore.isValid() && truststorePassword.isValid() && keystore.isValid()
-                && keystorePassword.isValid()) {
+        boolean configured = false;
+
+        if (truststore.isValid() && truststorePassword.isValid()) {
+            // ===== Truststore
 
-            addEnvVar(env, KAFKASQL_SECURITY_PROTOCOL, "SSL");
+            addEnvVar(env, KAFKASQL_SSL_TRUSTSTORE_TYPE, "PKCS12");
+            truststore.applySecretVolume(deployment, containerName);
+            addEnvVar(env, KAFKASQL_SSL_TRUSTSTORE_LOCATION, truststore.getSecretVolumeKeyPath());
+            truststorePassword.applySecretEnvVar(env, KAFKASQL_SSL_TRUSTSTORE_PASSWORD);
+
+            configured = true;
+        }
 
+        if (keystore.isValid()
+                && keystorePassword.isValid()) {
             // ===== Keystore
 
             addEnvVar(env, KAFKASQL_SSL_KEYSTORE_TYPE, "PKCS12");
             keystore.applySecretVolume(deployment, containerName);
             addEnvVar(env, KAFKASQL_SSL_KEYSTORE_LOCATION, keystore.getSecretVolumeKeyPath());
             keystorePassword.applySecretEnvVar(env, KAFKASQL_SSL_KEYSTORE_PASSWORD);
 
-            // ===== Truststore
-
-            addEnvVar(env, KAFKASQL_SSL_TRUSTSTORE_TYPE, "PKCS12");
-            truststore.applySecretVolume(deployment, containerName);
-            addEnvVar(env, KAFKASQL_SSL_TRUSTSTORE_LOCATION, truststore.getSecretVolumeKeyPath());
-            truststorePassword.applySecretEnvVar(env, KAFKASQL_SSL_TRUSTSTORE_PASSWORD);
-
-            return true;
+            configured = true;
         }
-        return false;
+
+        return configured;
     }
 
     private static Optional<KafkaSqlTLSSpec> getKafkaSqlTLSSpec(ApicurioRegistry3 primary) {

operator/controller/src/main/java/io/apicurio/registry/operator/feat/TLS.java

@@ -0,0 +1,67 @@
+package io.apicurio.registry.operator.feat;
+
+import io.apicurio.registry.operator.api.v1.ApicurioRegistry3;
+import io.apicurio.registry.operator.api.v1.ApicurioRegistry3Spec;
+import io.apicurio.registry.operator.api.v1.spec.AppSpec;
+import io.apicurio.registry.operator.api.v1.spec.TLSSpec;
+import io.apicurio.registry.operator.utils.SecretKeyRefTool;
+import io.fabric8.kubernetes.api.model.EnvVar;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+
+import java.util.Map;
+import java.util.Optional;
+
+import static io.apicurio.registry.operator.EnvironmentVariables.*;
+import static io.apicurio.registry.operator.resource.app.AppDeploymentResource.addEnvVar;
+import static java.util.Optional.ofNullable;
+
+public class TLS {
+
+    public static void configureTLS(ApicurioRegistry3 primary, Deployment deployment,
+                               String containerName, Map<String, EnvVar> env) {
+
+        addEnvVar(env, QUARKUS_HTTP_INSECURE_REQUESTS, Optional.ofNullable(primary.getSpec())
+                .map(ApicurioRegistry3Spec::getApp)
+                .map(AppSpec::getTls)
+                .map(TLSSpec::getInsecureRequests)
+                .orElse("enabled"));
+
+        var keystore = new SecretKeyRefTool(getTlsSpec(primary)
+                .map(TLSSpec::getKeystoreSecretRef)
+                .orElse(null), "user.p12");
+
+        var keystorePassword = new SecretKeyRefTool(getTlsSpec(primary)
+                .map(TLSSpec::getKeystorePasswordSecretRef)
+                .orElse(null), "user.password");
+
+        var truststore = new SecretKeyRefTool(getTlsSpec(primary)
+                .map(TLSSpec::getTruststoreSecretRef)
+                .orElse(null), "ca.p12");
+
+        var truststorePassword = new SecretKeyRefTool(getTlsSpec(primary)
+                .map(TLSSpec::getTruststorePasswordSecretRef)
+                .orElse(null), "ca.password");
+
+        if (truststore.isValid() && truststorePassword.isValid()) {
+            // ===== Truststore
+            truststore.applySecretVolume(deployment, containerName);
+            addEnvVar(env, QUARKUS_TLS_TRUST_STORE_P12_PATH, truststore.getSecretVolumeKeyPath());
+            truststorePassword.applySecretEnvVar(env, QUARKUS_TLS_TRUST_STORE_P12_PASSWORD);
+        }
+
+        if (keystore.isValid()
+                && keystorePassword.isValid()) {
+            // ===== Keystore
+            keystore.applySecretVolume(deployment, containerName);
+            addEnvVar(env, QUARKUS_TLS_KEY_STORE_P12_PATH, keystore.getSecretVolumeKeyPath());
+            keystorePassword.applySecretEnvVar(env, QUARKUS_TLS_KEY_STORE_P12_PASSWORD);
+        }
+    }
+
+    private static Optional<TLSSpec> getTlsSpec(ApicurioRegistry3 primary) {
+        return ofNullable(primary)
+                .map(ApicurioRegistry3::getSpec)
+                .map(ApicurioRegistry3Spec::getApp)
+                .map(AppSpec::getTls);
+    }
+}

operator/controller/src/main/java/io/apicurio/registry/operator/resource/ResourceFactory.java

@@ -7,6 +7,7 @@
 import io.apicurio.registry.operator.api.v1.ApicurioRegistry3Spec;
 import io.apicurio.registry.operator.api.v1.spec.AppSpec;
 import io.apicurio.registry.operator.api.v1.spec.StudioUiSpec;
+import io.apicurio.registry.operator.api.v1.spec.TLSSpec;
 import io.apicurio.registry.operator.api.v1.spec.UiSpec;
 import io.apicurio.registry.operator.status.ValidationErrorConditionManager;
 import io.apicurio.registry.operator.status.StatusManager;
@@ -23,7 +24,7 @@
 import java.util.Map;
 import java.util.Optional;
 
-import static io.apicurio.registry.operator.Constants.DEFAULT_REPLICAS;
+import static io.apicurio.registry.operator.Constants.*;
 import static io.apicurio.registry.operator.api.v1.ContainerNames.*;
 import static io.apicurio.registry.operator.resource.Labels.getSelectorLabels;
 import static io.apicurio.registry.operator.resource.app.AppDeploymentResource.getContainerFromPodTemplateSpec;
@@ -56,6 +57,19 @@ public Deployment getDefaultAppDeployment(ApicurioRegistry3 primary) {
                         .map(AppSpec::getReplicas).orElse(DEFAULT_REPLICAS),
                 ofNullable(primary.getSpec()).map(ApicurioRegistry3Spec::getApp)
                         .map(AppSpec::getPodTemplateSpec).orElse(null)); // TODO:
+
+        var readinessProbe = new ProbeBuilder().withHttpGet(new HTTPGetActionBuilder().withPath("/health/ready").withPort(new IntOrString(8080)).withScheme("HTTP").build()).build();
+        var livenessProbe = new ProbeBuilder().withHttpGet(new HTTPGetActionBuilder().withPath("/health/live").withPort(new IntOrString(8080)).withScheme("HTTP").build()).build();
+
+        Optional<TLSSpec> tlsSpec = ofNullable(primary.getSpec())
+                .map(ApicurioRegistry3Spec::getApp)
+                .map(AppSpec::getTls);
+
+        if (tlsSpec.isPresent()) {
+            readinessProbe = TLS_DEFAULT_READINESS_PROBE;
+            livenessProbe = TLS_DEFAULT_LIVENESS_PROBE;
+        }
+
         // Replicas
         mergeDeploymentPodTemplateSpec(
                 COMPONENT_APP_SPEC_FIELD_NAME,
@@ -64,11 +78,12 @@ public Deployment getDefaultAppDeployment(ApicurioRegistry3 primary) {
                 REGISTRY_APP_CONTAINER_NAME,
                 Configuration.getAppImage(),
                 List.of(new ContainerPortBuilder().withName("http").withProtocol("TCP").withContainerPort(8080).build()),
-                new ProbeBuilder().withHttpGet(new HTTPGetActionBuilder().withPath("/health/ready").withPort(new IntOrString(8080)).withScheme("HTTP").build()).build(),
-                new ProbeBuilder().withHttpGet(new HTTPGetActionBuilder().withPath("/health/live").withPort(new IntOrString(8080)).withScheme("HTTP").build()).build(),
+                readinessProbe,
+                livenessProbe,
                 Map.of("cpu", new Quantity("500m"), "memory", new Quantity("512Mi")),
                 Map.of("cpu", new Quantity("1"), "memory", new Quantity("1Gi"))
         );
+
         addDefaultLabels(r.getMetadata().getLabels(), primary, COMPONENT_APP);
         addSelectorLabels(r.getSpec().getSelector().getMatchLabels(), primary, COMPONENT_APP);
         addDefaultLabels(r.getSpec().getTemplate().getMetadata().getLabels(), primary, COMPONENT_APP);

operator/controller/src/main/java/io/apicurio/registry/operator/resource/app/AppDeploymentResource.java

@@ -11,6 +11,7 @@
 import io.apicurio.registry.operator.feat.Cors;
 import io.apicurio.registry.operator.feat.KafkaSql;
 import io.apicurio.registry.operator.feat.PostgresSql;
+import io.apicurio.registry.operator.feat.TLS;
 import io.apicurio.registry.operator.feat.security.Auth;
 import io.apicurio.registry.operator.status.ReadyConditionManager;
 import io.apicurio.registry.operator.status.StatusManager;
@@ -87,6 +88,9 @@ protected Deployment desired(ApicurioRegistry3 primary, Context<ApicurioRegistry
         // Configure the CORS_ALLOWED_ORIGINS env var based on the ingress host
         Cors.configureAllowedOrigins(primary, envVars);
 
+        // Configure the TLS env vars
+        TLS.configureTLS(primary, deployment, REGISTRY_APP_CONTAINER_NAME, envVars);
+
         // Enable the "mutability" feature in Registry, but only if Studio is deployed. It is based on Service
         // in case a custom Ingress is used.
         var sOpt = context.getSecondaryResource(STUDIO_UI_SERVICE_KEY.getKlass(),