Add 'mirrord-operator-secret-access' role to operator (#2960)

* Add 'mirrord-operator-secret-access' role to operator

* Rename operator roles and bindings, and add RoleBinding

* Add namespace to role binding

changelog.d/+kafka-secret-permission.internal.md

@@ -0,0 +1 @@
+Allow the operator to fetch Secrets in the operator namespace with 'mirrord-operator-secret-access' role

mirrord/operator/src/setup.rs

@@ -37,6 +37,8 @@ pub static OPERATOR_NAME: &str = "mirrord-operator";
 static OPERATOR_PORT: i32 = 443;
 static OPERATOR_ROLE_NAME: &str = "mirrord-operator";
 static OPERATOR_ROLE_BINDING_NAME: &str = "mirrord-operator";
+static OPERATOR_CLUSTER_ROLE_NAME: &str = "mirrord-operator";
+static OPERATOR_CLUSTER_ROLE_BINDING_NAME: &str = "mirrord-operator";
 static OPERATOR_CLIENT_CA_ROLE_NAME: &str = "mirrord-operator-apiserver-authentication";
 static OPERATOR_CLUSTER_USER_ROLE_NAME: &str = "mirrord-operator-user";
 static OPERATOR_LICENSE_SECRET_NAME: &str = "mirrord-operator-license";
@@ -103,8 +105,10 @@ pub struct Operator {
     deployment: OperatorDeployment,
     license_secret: Option<OperatorLicenseSecret>,
     namespace: OperatorNamespace,
-    role: OperatorRole,
-    role_binding: OperatorRoleBinding,
+    cluster_role: OperatorClusterRole,
+    cluster_role_binding: OperatorClusterRoleBinding,
+    role: Option<OperatorRole>,
+    role_binding: Option<OperatorRoleBinding>,
     service: OperatorService,
     service_account: OperatorServiceAccount,
     user_cluster_role: OperatorClusterUserRole,
@@ -135,10 +139,19 @@ impl Operator {
 
         let service_account = OperatorServiceAccount::new(&namespace, aws_role_arn);
 
-        let role = OperatorRole::new(sqs_splitting, kafka_splitting, application_auto_pause);
-        let role_binding = OperatorRoleBinding::new(&role, &service_account);
+        let cluster_role =
+            OperatorClusterRole::new(sqs_splitting, kafka_splitting, application_auto_pause);
+        let cluster_role_binding = OperatorClusterRoleBinding::new(&cluster_role, &service_account);
         let user_cluster_role = OperatorClusterUserRole::new();
 
+        let (role, role_binding) = kafka_splitting
+            .then(|| {
+                let role = OperatorRole::new(&namespace);
+                let role_binding = OperatorRoleBinding::new(&role, &service_account, &namespace);
+                (role, role_binding)
+            })
+            .unzip();
+
         let client_ca_role = OperatorClientCaRole::new();
         let client_ca_role_binding =
             OperatorClientCaRoleBinding::new(&client_ca_role, &service_account);
@@ -163,6 +176,8 @@ impl Operator {
             deployment,
             license_secret,
             namespace,
+            cluster_role,
+            cluster_role_binding,
             role,
             role_binding,
             service,
@@ -189,7 +204,7 @@ impl OperatorSetup for Operator {
         self.service_account.to_writer(&mut writer)?;
 
         writer.write_all(b"---\n")?;
-        self.role.to_writer(&mut writer)?;
+        self.cluster_role.to_writer(&mut writer)?;
 
         writer.write_all(b"---\n")?;
         self.user_cluster_role.to_writer(&mut writer)?;
@@ -198,7 +213,7 @@ impl OperatorSetup for Operator {
         self.client_ca_role.to_writer(&mut writer)?;
 
         writer.write_all(b"---\n")?;
-        self.role_binding.to_writer(&mut writer)?;
+        self.cluster_role_binding.to_writer(&mut writer)?;
 
         writer.write_all(b"---\n")?;
         self.client_ca_role_binding.to_writer(&mut writer)?;
@@ -232,6 +247,16 @@ impl OperatorSetup for Operator {
 
             writer.write_all(b"---\n")?;
             MirrordKafkaTopicsConsumer::crd().to_writer(&mut writer)?;
+
+            if let Some(role) = self.role.as_ref() {
+                writer.write_all(b"---\n")?;
+                role.to_writer(&mut writer)?;
+            }
+
+            if let Some(role_binding) = self.role_binding.as_ref() {
+                writer.write_all(b"---\n")?;
+                role_binding.to_writer(&mut writer)?;
+            }
         }
 
         Ok(())
@@ -475,9 +500,9 @@ impl OperatorServiceAccount {
 }
 
 #[derive(Debug)]
-pub struct OperatorRole(ClusterRole);
+pub struct OperatorClusterRole(ClusterRole);
 
-impl OperatorRole {
+impl OperatorClusterRole {
     pub fn new(sqs_splitting: bool, kafka_splitting: bool, application_auto_pause: bool) -> Self {
         let mut rules = vec![
             PolicyRule {
@@ -648,14 +673,14 @@ impl OperatorRole {
 
         let role = ClusterRole {
             metadata: ObjectMeta {
-                name: Some(OPERATOR_ROLE_NAME.to_owned()),
+                name: Some(OPERATOR_CLUSTER_ROLE_NAME.to_owned()),
                 ..Default::default()
             },
             rules: Some(rules),
             ..Default::default()
         };
 
-        OperatorRole(role)
+        OperatorClusterRole(role)
     }
 
     fn as_role_ref(&self) -> RoleRef {
@@ -667,20 +692,82 @@ impl OperatorRole {
     }
 }
 
-impl Default for OperatorRole {
+impl Default for OperatorClusterRole {
     fn default() -> Self {
         Self::new(false, false, false)
     }
 }
 
 #[derive(Debug)]
-pub struct OperatorRoleBinding(ClusterRoleBinding);
+pub struct OperatorClusterRoleBinding(ClusterRoleBinding);
 
-impl OperatorRoleBinding {
-    pub fn new(role: &OperatorRole, sa: &OperatorServiceAccount) -> Self {
+impl OperatorClusterRoleBinding {
+    pub fn new(role: &OperatorClusterRole, sa: &OperatorServiceAccount) -> Self {
         let role_binding = ClusterRoleBinding {
+            metadata: ObjectMeta {
+                name: Some(OPERATOR_CLUSTER_ROLE_BINDING_NAME.to_owned()),
+                ..Default::default()
+            },
+            role_ref: role.as_role_ref(),
+            subjects: Some(vec![sa.as_subject()]),
+        };
+
+        OperatorClusterRoleBinding(role_binding)
+    }
+}
+
+#[derive(Debug)]
+pub struct OperatorRole(Role);
+
+impl OperatorRole {
+    pub fn new(namespace: &OperatorNamespace) -> Self {
+        let rules = vec![
+            // Allow the operator to fetch Secrets in the operator's namespace
+            PolicyRule {
+                api_groups: Some(vec![MirrordKafkaClientConfig::group(&()).into_owned()]),
+                resources: Some(vec!["secrets".to_owned()]),
+                verbs: ["get", "list", "watch"]
+                    .into_iter()
+                    .map(String::from)
+                    .collect(),
+                ..Default::default()
+            },
+        ];
+
+        let role = Role {
+            metadata: ObjectMeta {
+                name: Some(OPERATOR_ROLE_NAME.to_owned()),
+                namespace: Some(namespace.name().to_owned()),
+                ..Default::default()
+            },
+            rules: Some(rules),
+        };
+
+        Self(role)
+    }
+
+    fn as_role_ref(&self) -> RoleRef {
+        RoleRef {
+            api_group: "rbac.authorization.k8s.io".to_owned(),
+            kind: "Role".to_owned(),
+            name: self.0.metadata.name.clone().unwrap_or_default(),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct OperatorRoleBinding(RoleBinding);
+
+impl OperatorRoleBinding {
+    pub fn new(
+        role: &OperatorRole,
+        sa: &OperatorServiceAccount,
+        namespace: &OperatorNamespace,
+    ) -> Self {
+        let role_binding = RoleBinding {
             metadata: ObjectMeta {
                 name: Some(OPERATOR_ROLE_BINDING_NAME.to_owned()),
+                namespace: Some(namespace.name().to_owned()),
                 ..Default::default()
             },
             role_ref: role.as_role_ref(),
@@ -909,12 +996,14 @@ writer_impl![
     OperatorNamespace,
     OperatorDeployment,
     OperatorServiceAccount,
-    OperatorRole,
-    OperatorRoleBinding,
+    OperatorClusterRole,
+    OperatorClusterRoleBinding,
     OperatorLicenseSecret,
     OperatorService,
     OperatorApiService,
     OperatorClusterUserRole,
     OperatorClientCaRole,
-    OperatorClientCaRoleBinding
+    OperatorClientCaRoleBinding,
+    OperatorRole,
+    OperatorRoleBinding
 ];