fix: do not pass catalog credentials through the environment

rust/operator-binary/src/catalog/commons.rs

@@ -123,8 +123,8 @@ impl ExtendCatalogConfig for s3::v1alpha1::InlineConnectionOrReference {
                 ..=468 => ("hive.s3.aws-access-key", "hive.s3.aws-secret-key"),
                 469.. => ("s3.aws-access-key", "s3.aws-secret-key"),
             };
-            catalog_config.add_env_property_from_file(access_key_prop, access_key);
-            catalog_config.add_env_property_from_file(secret_key_prop, secret_key);
+            catalog_config.add_property_from_file(access_key_prop, access_key);
+            catalog_config.add_property_from_file(secret_key_prop, secret_key);
         }
 
         match trino_version {

rust/operator-binary/src/catalog/config.rs

@@ -21,11 +21,6 @@ pub struct CatalogConfig {
     /// List of EnvVar that will be added to every Trino container
     pub env_bindings: Vec<EnvVar>,
 
-    /// Env-Vars that should be exported.
-    /// The value will be read from the file specified.
-    /// You can think of it like `export <key>="$(cat <value>)"`
-    pub load_env_from_files: BTreeMap<String, String>,
-
     /// Additional commands that needs to be executed before starting Trino
     pub init_container_extra_start_commands: Vec<String>,
 
@@ -42,7 +37,6 @@ impl CatalogConfig {
             name: name.into(),
             properties: BTreeMap::new(),
             env_bindings: Vec::new(),
-            load_env_from_files: BTreeMap::new(),
             init_container_extra_start_commands: Vec::new(),
             volumes: Vec::new(),
             volume_mounts: Vec::new(),
@@ -55,15 +49,14 @@ impl CatalogConfig {
         self.properties.insert(property.into(), value.into());
     }
 
-    pub fn add_env_property_from_file(
+    pub fn add_property_from_file(
         &mut self,
         property: impl Into<String>,
         file_name: impl Into<String>,
     ) {
         let property = property.into();
-        let env_name = calculate_env_name(&self.name, &property);
-        self.add_property(&property, format!("${{ENV:{env_name}}}"));
-        self.load_env_from_files.insert(env_name, file_name.into());
+        let file = file_name.into();
+        self.add_property(&property, format!("${{file:UTF-8:{file}}}"));
     }
 
     pub fn add_env_property_from_secret(

rust/operator-binary/src/command.rs

@@ -82,10 +82,7 @@ pub fn container_prepare_args(
     args
 }
 
-pub fn container_trino_args(
-    authentication_config: &TrinoAuthenticationConfig,
-    catalogs: &[CatalogConfig],
-) -> Vec<String> {
+pub fn container_trino_args(authentication_config: &TrinoAuthenticationConfig) -> Vec<String> {
     let mut args = vec![
         // copy config files to a writeable empty folder
         format!(
@@ -106,11 +103,12 @@ pub fn container_trino_args(
     // Add the commands that are needed to set up the catalogs
     // Don't print secret contents!
     args.push("set +x".to_string());
-    catalogs.iter().for_each(|catalog| {
-        for (env_name, file) in &catalog.load_env_from_files {
-            args.push(format!("export {env_name}=\"$(cat {file})\""));
-        }
-    });
+
+    // Resolve credentials in all catalog configs.
+    args.push(format!(
+        "find {rw_conf}/catalog -type f -exec config-utils template '{{}}' ';'",
+        rw_conf = RW_CONFIG_DIR_NAME
+    ));
 
     // Resolve credentials for fault tolerant execution exchange manager if needed
     args.push(format!(

rust/operator-binary/src/controller.rs

@@ -1243,7 +1243,7 @@ fn build_rolegroup_statefulset(
             "-c".to_string(),
         ])
         .args(vec![
-            command::container_trino_args(trino_authentication_config, catalogs).join("\n"),
+            command::container_trino_args(trino_authentication_config).join("\n"),
         ])
         .add_env_vars(env)
         .add_volume_mount("config", CONFIG_DIR_NAME)

tests/templates/kuttl/opa-authorization/20-install-trino.yaml.j2

@@ -1,4 +1,48 @@
 ---
+apiVersion: trino.stackable.tech/v1alpha1
+kind: TrinoCluster
+metadata:
+  name: trino
+spec:
+  image:
+{% if test_scenario['values']['trino'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['trino'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['trino'].split(',')[0] }}"
+{% else %}
+    productVersion: "{{ test_scenario['values']['trino'] }}"
+{% endif %}
+    pullPolicy: IfNotPresent
+  clusterConfig:
+    catalogLabelSelector:
+      matchLabels:
+        trino: trino
+    authentication:
+      - authenticationClass: trino-users-auth
+    authorization:
+      opa:
+        configMapName: opa
+        package: trino
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+    vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  coordinators:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+    roleGroups:
+      default:
+        replicas: 1
+        config: {}
+  workers:
+    config:
+      gracefulShutdownTimeout: 10s # Let the test run faster
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+    roleGroups:
+      default:
+        replicas: 1
+        config: {}
+---
 apiVersion: authentication.stackable.tech/v1alpha1
 kind: AuthenticationClass
 metadata:
@@ -64,47 +108,3 @@ spec:
         configMap: hive # It's fine to reuse the existing HMS for tests. Not recommended for production though, there a dedicated HMS should be used.
       s3:
         reference: minio
----
-apiVersion: trino.stackable.tech/v1alpha1
-kind: TrinoCluster
-metadata:
-  name: trino
-spec:
-  image:
-{% if test_scenario['values']['trino'].find(",") > 0 %}
-    custom: "{{ test_scenario['values']['trino'].split(',')[1] }}"
-    productVersion: "{{ test_scenario['values']['trino'].split(',')[0] }}"
-{% else %}
-    productVersion: "{{ test_scenario['values']['trino'] }}"
-{% endif %}
-    pullPolicy: IfNotPresent
-  clusterConfig:
-    catalogLabelSelector:
-      matchLabels:
-        trino: trino
-    authentication:
-      - authenticationClass: trino-users-auth
-    authorization:
-      opa:
-        configMapName: opa
-        package: trino
-{% if lookup('env', 'VECTOR_AGGREGATOR') %}
-    vectorAggregatorConfigMapName: vector-aggregator-discovery
-{% endif %}
-  coordinators:
-    config:
-      logging:
-        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
-    roleGroups:
-      default:
-        replicas: 1
-        config: {}
-  workers:
-    config:
-      gracefulShutdownTimeout: 10s # Let the test run faster
-      logging:
-        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
-    roleGroups:
-      default:
-        replicas: 1
-        config: {}