Implement multi-level authentication on pulumi-operator

This commit adds functionality for multi-level authentication in the `pulumi_operator`. In the `pulumi_execution.rs` file, code changes are made to clone the namespace and retrieve secret keys such as `AWS_ACCESS_KEY_ID`, `AWS_DEFAULT_REGION` and `AWS_SECRET_ACCESS_KEY` from the data.

The changes also include setting environment variables with these secret keys to be used later for authentication. These improvements grant more security and efficiency in pulumi execution tasks on Kubernetes platform.

In the `Cargo.toml` file, modification has been done to include `install-crds` as a feature. The `main.rs` in `pulumi_operator_kubernetes` now creates a service account, role and role binding for each stack. This increases granularity in access control.

Lastly, in `crd.rs` additional configurations for the main container has been added. This provides the ability to add extra environment variables and volumes, giving the main container more flexibility.

pulumi-operator-kubernetes-job/src/pulumi_execution.rs

@@ -64,7 +64,7 @@ impl PulumiExecution {
         String::from_utf8(
           self
             .kubernetes_service
-            .get_in_namespace::<Secret>(namespace, secret_name)
+            .get_in_namespace::<Secret>(namespace.clone(), secret_name)
             .await
             .unwrap()
             .data
@@ -78,8 +78,31 @@ impl PulumiExecution {
       ),
     };
 
+    if let Some(secret_name) = &inner_stack_auth.backend_auth_secret {
+      let data = self
+        .kubernetes_service
+        .get_in_namespace::<Secret>(namespace, secret_name)
+        .await
+        .unwrap()
+        .data
+        .unwrap();
+
+      let access_key_id =
+        String::from_utf8(data.get("AWS_ACCESS_KEY_ID").unwrap().0.clone())
+          .unwrap();
+      let default_region =
+        String::from_utf8(data.get("AWS_DEFAULT_REGION").unwrap().0.clone())
+          .unwrap();
+      let secret_access_key =
+        String::from_utf8(data.get("AWS_SECRET_ACCESS_KEY").unwrap().0.clone())
+          .unwrap();
+
+      std::env::set_var("AWS_ACCESS_KEY_ID", access_key_id);
+      std::env::set_var("AWS_DEFAULT_REGION", default_region);
+      std::env::set_var("AWS_SECRET_ACCESS_KEY", secret_access_key);
+    }
+
     if let Some(access_token) = access_token {
-      dbg!(&access_token);
       std::env::set_var("PULUMI_CONFIG_PASSPHRASE", access_token);
     }
 
@@ -119,7 +142,7 @@ impl PulumiExecution {
 
     pulumi
       .login(LoginOptions {
-        url: "gs://stromee-pulumi".to_string(),
+        url: inner_stack_auth.backend,
       })
       .await;
 

pulumi-operator-kubernetes/Cargo.toml

@@ -22,4 +22,4 @@ warp = "0.3.5"
 [features]
 install-crds = []
 boot = []
-default = ["boot"]
+default = ["boot", "install-crds"]

pulumi-operator-kubernetes/src/stack/auth/inner.rs

@@ -5,5 +5,6 @@ use serde::{Deserialize, Serialize};
 #[serde(rename_all = "camelCase")]
 pub struct InnerStackAuthSpec {
   pub backend: String,
+  pub backend_auth_secret: Option<String>,
   pub access_token_secret: Option<String>,
-}
+}

pulumi-operator-kubernetes/src/stack/controller_strategy.rs

@@ -151,7 +151,6 @@ impl KubernetesPulumiStackControllerStrategy {
     // self.start_admission_controller().await?;
 
     *self.controller_stream.lock().await = Some(Box::pin(controller) as _);
-    println!("sososo");
     Ok(())
   }
 

pulumi-operator-kubernetes/src/stack/crd.rs

@@ -1,4 +1,4 @@
-use k8s_openapi::api::core::v1::Container;
+use k8s_openapi::api::core::v1::{Container, EnvVar, Volume, VolumeMount};
 use k8s_openapi::schemars::JsonSchema;
 use kube::CustomResource;
 use serde::{Deserialize, Serialize};
@@ -21,6 +21,15 @@ pub struct StackSpec {
   pub auth: StackAuthRef,
   pub path: Option<String>,
   pub init_containers: Option<Vec<Container>>,
+  pub extra_volumes: Option<Vec<Volume>>,
+  pub main_container: Option<MainContainerOverride>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct MainContainerOverride {
+  pub extra_volume_mounts: Option<Vec<VolumeMount>>,
+  pub extra_env: Option<Vec<EnvVar>>,
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, JsonSchema)]

pulumi-operator-kubernetes/src/stack/service.rs

@@ -1,7 +1,10 @@
 use async_trait::async_trait;
 use futures::{StreamExt, TryStreamExt};
 use k8s_openapi::api::batch::v1::Job;
-use kube::api::{DeleteParams, PostParams, WatchEvent};
+use k8s_openapi::api::core::v1::{Container, ServiceAccount};
+use k8s_openapi::api::rbac::v1::{Role, RoleBinding};
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+use kube::api::{DeleteParams, Object, PostParams, WatchEvent};
 use serde_json::json;
 use springtime_di::{component_alias, Component};
 use std::error::Error;
@@ -38,21 +41,63 @@ impl KubernetesPulumiStackService {
     stack: PulumiStack,
   ) -> Result<(), PulumiStackServiceError> {
     self.cancel_stack(stack.clone()).await?;
+    self.create_service_account(stack.clone()).await?;
+    self.create_role(stack.clone()).await?;
+    self.create_role_binding(stack.clone()).await?;
+
     let name = stack.metadata.name.unwrap();
     let namespace = stack.metadata.namespace.unwrap();
     let init_containers = stack.spec.init_containers;
+    let extra_volumes = stack.spec.extra_volumes;
+    let container_override = stack.spec.main_container;
 
     let operator_namespace = self
       .config_provider
       .operator_namespace()
       .map_err(|e| PulumiStackServiceError::Config(Box::new(e)))?; // TODO
 
+    let mut main_container: Container = serde_json::from_value(json!({
+        "name": "pulumi",
+        "image": "ghcr.io/stromee/pulumi-operator/pulumi-operator-kubernetes-job:1.0.16",
+        "env": [{
+            "name": "PULUMI_STACK",
+            "value": name
+        }, {
+            "name": "WATCH_NAMESPACE",
+            "value": namespace
+        }, {
+            "name": ConfigProvider::OPERATOR_NS_VAR,
+            "value": operator_namespace
+        }, {
+            "name": "RUST_BACKTRACE",
+            "value": "full"
+        }],
+        "imagePullPolicy": "Always"
+    })).unwrap();
+
+    if let Some(main_container_override) = container_override {
+      if let Some(mut extra_volume_mounts) =
+        main_container_override.extra_volume_mounts
+      {
+        let mut volume_mounts =
+          main_container.volume_mounts.clone().unwrap_or_default();
+        volume_mounts.append(&mut extra_volume_mounts);
+        main_container.volume_mounts.replace(volume_mounts);
+      }
+
+      if let Some(mut extra_env) = main_container_override.extra_env {
+        let mut env = main_container.env.clone().unwrap_or_default();
+        env.append(&mut extra_env);
+        main_container.env.replace(env);
+      }
+    }
+
     let job = serde_json::from_value(json!({
         "apiVersion": "batch/v1",
         "kind": "Job",
         "metadata": {
             "name": name,
-            "namespace": "pulumi-operator"
+            "namespace": namespace.clone()
         },
         "spec": {
             "template": {
@@ -61,25 +106,9 @@ impl KubernetesPulumiStackService {
                 },
                 "spec": {
                     "initContainers": init_containers,
-                    "containers": [{
-                        "name": "pulumi",
-                        "image": "ghcr.io/stromee/pulumi-operator/pulumi-operator-kubernetes-job:1.0.15",
-                        "env": [{
-                            "name": "PULUMI_STACK",
-                            "value": name
-                        }, {
-                            "name": "WATCH_NAMESPACE",
-                            "value": namespace
-                        }, {
-                            "name": ConfigProvider::OPERATOR_NS_VAR,
-                            "value": operator_namespace
-                        }, {
-                            "name": "RUST_BACKTRACE",
-                            "value": "full"
-                        }],
-                        "imagePullPolicy": "Always"
-                    }],
-                    "serviceAccountName": "superuser",
+                    "containers": [main_container],
+                    "volumes": extra_volumes,
+                    "serviceAccountName": &name,
                     "restartPolicy": "Never"
                 }
             },
@@ -92,18 +121,145 @@ impl KubernetesPulumiStackService {
 
     let api = self
       .kubernetes_service
-      .all_in_namespace_api::<Job>("pulumi-operator")
+      .all_in_namespace_api::<Job>(namespace.clone())
       .await;
 
     api
       .create(&PostParams::default(), &job)
       .await
       .map_err(|err| PulumiStackServiceError::UpdateFailed(err.into()))?;
-    dbg!("Update stack123");
 
     Ok(())
   }
 
+  async fn create_service_account(
+    &self,
+    stack: PulumiStack,
+  ) -> Result<(), PulumiStackServiceError> {
+    let namespace = stack.metadata.namespace.unwrap();
+    let name = stack.metadata.name.clone().unwrap();
+
+    let service_account: ServiceAccount = serde_json::from_value(json!({
+        "apiVersion": "v1",
+        "kind": "ServiceAccount",
+        "metadata": {
+            "name": &name,
+            "namespace": &namespace
+        }
+    }))
+    .unwrap();
+
+    let api = self
+      .kubernetes_service
+      .all_in_namespace_api(&namespace)
+      .await;
+    match api.get(&name).await {
+      Ok(_) => {
+        api
+          .replace(&name, &PostParams::default(), &service_account)
+          .await
+          .unwrap();
+      }
+      Err(_) => {
+        api
+          .create(&PostParams::default(), &service_account)
+          .await
+          .unwrap();
+      }
+    }
+    Ok(())
+  }
+
+  async fn create_role(
+    &self,
+    stack: PulumiStack,
+  ) -> Result<(), PulumiStackServiceError> {
+    let namespace = stack.metadata.namespace.unwrap();
+    let name = stack.metadata.name.clone().unwrap();
+
+    let role: Role = serde_json::from_value(json!({
+        "apiVersion": "rbac.authorization.k8s.io/v1",
+        "kind": "Role",
+        "metadata": {
+            "name": name,
+            "namespace": namespace
+        },
+        "rules": [{
+            "apiGroups": ["*"],
+            "resources": ["*"],
+            "verbs": ["get", "list", "watch"]
+        }]
+    }))
+    .unwrap();
+
+    let api = self
+      .kubernetes_service
+      .all_in_namespace_api(&namespace)
+      .await;
+
+    match api.get(&name).await {
+      Ok(_) => {
+        api
+          .replace(&name, &PostParams::default(), &role)
+          .await
+          .unwrap();
+      }
+      Err(_) => {
+        api.create(&PostParams::default(), &role).await.unwrap();
+      }
+    }
+    Ok(())
+  }
+
+  async fn create_role_binding(
+    &self,
+    stack: PulumiStack,
+  ) -> Result<(), PulumiStackServiceError> {
+    let namespace = stack.metadata.namespace.unwrap();
+    let name = stack.metadata.name.clone().unwrap();
+
+    let role_binding: RoleBinding = serde_json::from_value(json!({
+        "apiVersion": "rbac.authorization.k8s.io/v1",
+        "kind": "RoleBinding",
+        "metadata": {
+            "name": name,
+            "namespace": namespace
+        },
+        "subjects": [{
+            "kind": "ServiceAccount",
+            "name": name,
+            "namespace": namespace
+        }],
+        "roleRef": {
+            "apiGroup": "rbac.authorization.k8s.io",
+            "kind": "Role",
+            "name": name
+        }
+    }))
+    .unwrap();
+
+    let api = self
+      .kubernetes_service
+      .all_in_namespace_api(&namespace)
+      .await;
+
+    match api.get(&name).await {
+      Ok(_) => {
+        api
+          .replace(&name, &PostParams::default(), &role_binding)
+          .await
+          .unwrap();
+      }
+      Err(_) => {
+        api
+          .create(&PostParams::default(), &role_binding)
+          .await
+          .unwrap();
+      }
+    }
+
+    Ok(())
+  }
   pub(crate) async fn cancel_stack(
     &self,
     stack: PulumiStack,
@@ -112,7 +268,7 @@ impl KubernetesPulumiStackService {
     let name = stack.metadata.name.unwrap();
     let api = self
       .kubernetes_service
-      .all_in_namespace_api::<Job>("pulumi-operator")
+      .all_in_namespace_api::<Job>(namespace.clone())
       .await;
 
     if api.get(&name).await.is_err() {