fix: Refuse to overwrite foreign objects from TrustStore reconciler

Author: dervoeti

rust/operator-binary/src/truststore_controller.rs

@@ -8,6 +8,7 @@ use stackable_operator::{
     k8s_openapi::{
         ByteString,
         api::core::v1::{ConfigMap, Secret},
+        apimachinery::pkg::apis::meta::v1::OwnerReference,
     },
     kube::{
         Resource,
@@ -220,6 +221,27 @@ pub enum Error {
         source: stackable_operator::client::Error,
         secret: ObjectRef<Secret>,
     },
+
+    #[snafu(display(
+        "failed to look up pre-existing target {object} before applying the TrustStore"
+    ))]
+    GetExistingTarget {
+        source: stackable_operator::client::Error,
+        object: ObjectRef<stackable_operator::kube::api::DynamicObject>,
+    },
+
+    #[snafu(display(
+        "refusing to overwrite pre-existing {object} that is not owned by this TrustStore"
+    ))]
+    RefuseToOverwriteForeignTarget {
+        object: ObjectRef<stackable_operator::kube::api::DynamicObject>,
+    },
+
+    #[snafu(display("TrustStore has no associated UID"))]
+    NoTrustStoreUid,
+
+    #[snafu(display("TrustStore has no associated name"))]
+    NoTrustStoreName,
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -239,10 +261,28 @@ impl ReconcilerError for Error {
             Error::BuildOwnerReference { .. } => None,
             Error::ApplyTrustStoreConfigMap { config_map, .. } => Some(config_map.clone().erase()),
             Error::ApplyTrustStoreSecret { secret, .. } => Some(secret.clone().erase()),
+            Error::GetExistingTarget { object, .. } => Some(object.clone()),
+            Error::RefuseToOverwriteForeignTarget { object } => Some(object.clone()),
+            Error::NoTrustStoreUid => None,
+            Error::NoTrustStoreName => None,
         }
     }
 }
 
+/// Returns `true` if `existing_owners` contain a controller `OwnerReference` that points to the
+/// TrustStore with the given UID. Used to ensure we only overwrite output ConfigMaps and Secrets
+/// that we previously created ourselves; refusing otherwise prevents the TrustStore primitive
+/// from being abused to clobber foreign same-named objects (e.g. `kube-root-ca.crt`) via the
+/// operator's elevated cluster-wide write permissions.
+fn is_owned_by_truststore(existing_owners: &[OwnerReference], truststore_uid: &str) -> bool {
+    existing_owners.iter().any(|owner| {
+        owner.controller == Some(true)
+            && owner.kind == "TrustStore"
+            && owner.api_version.starts_with("secrets.stackable.tech/")
+            && owner.uid == truststore_uid
+    })
+}
+
 struct Ctx {
     client: stackable_operator::client::Client,
 }
@@ -311,6 +351,18 @@ async fn reconcile(
         .context(BuildOwnerReferenceSnafu)?
         .build();
 
+    let truststore_name = truststore
+        .metadata
+        .name
+        .as_deref()
+        .context(NoTrustStoreNameSnafu)?;
+    let truststore_namespace = selector.namespace.as_str();
+    let truststore_uid = truststore
+        .metadata
+        .uid
+        .as_deref()
+        .context(NoTrustStoreUidSnafu)?;
+
     match truststore.spec.target_kind {
         v1alpha1::TrustStoreOutputType::ConfigMap => {
             let trust_cm = ConfigMap {
@@ -319,6 +371,22 @@ async fn reconcile(
                 binary_data: Some(binary_data),
                 ..Default::default()
             };
+            let existing = ctx
+                .client
+                .get_opt::<ConfigMap>(truststore_name, truststore_namespace)
+                .await
+                .with_context(|_| GetExistingTargetSnafu {
+                    object: ObjectRef::from_obj(&trust_cm).erase(),
+                })?;
+            if let Some(existing) = existing {
+                let owners = existing.metadata.owner_references.as_deref().unwrap_or(&[]);
+                if !is_owned_by_truststore(owners, truststore_uid) {
+                    return RefuseToOverwriteForeignTargetSnafu {
+                        object: ObjectRef::from_obj(&trust_cm).erase(),
+                    }
+                    .fail();
+                }
+            }
             ctx.client
                 .apply_patch(CONTROLLER_NAME, &trust_cm, &trust_cm)
                 .await
@@ -333,6 +401,22 @@ async fn reconcile(
                 data: Some(binary_data),
                 ..Default::default()
             };
+            let existing = ctx
+                .client
+                .get_opt::<Secret>(truststore_name, truststore_namespace)
+                .await
+                .with_context(|_| GetExistingTargetSnafu {
+                    object: ObjectRef::from_obj(&trust_secret).erase(),
+                })?;
+            if let Some(existing) = existing {
+                let owners = existing.metadata.owner_references.as_deref().unwrap_or(&[]);
+                if !is_owned_by_truststore(owners, truststore_uid) {
+                    return RefuseToOverwriteForeignTargetSnafu {
+                        object: ObjectRef::from_obj(&trust_secret).erase(),
+                    }
+                    .fail();
+                }
+            }
             ctx.client
                 .apply_patch(CONTROLLER_NAME, &trust_secret, &trust_secret)
                 .await
@@ -352,3 +436,84 @@ fn error_policy(
 ) -> controller::Action {
     controller::Action::requeue(Duration::from_secs(5))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn owner_ref(
+        api_version: &str,
+        kind: &str,
+        uid: &str,
+        controller: Option<bool>,
+    ) -> OwnerReference {
+        OwnerReference {
+            api_version: api_version.to_string(),
+            kind: kind.to_string(),
+            name: "some-name".to_string(),
+            uid: uid.to_string(),
+            controller,
+            block_owner_deletion: None,
+        }
+    }
+
+    #[test]
+    fn empty_owner_refs_are_rejected() {
+        assert!(!is_owned_by_truststore(&[], "my-uid"));
+    }
+
+    #[test]
+    fn matching_controller_owner_ref_is_accepted() {
+        let owners = [owner_ref(
+            "secrets.stackable.tech/v1alpha1",
+            "TrustStore",
+            "my-uid",
+            Some(true),
+        )];
+        assert!(is_owned_by_truststore(&owners, "my-uid"));
+    }
+
+    #[test]
+    fn non_controller_owner_ref_is_rejected() {
+        let owners = [owner_ref(
+            "secrets.stackable.tech/v1alpha1",
+            "TrustStore",
+            "my-uid",
+            Some(false),
+        )];
+        assert!(!is_owned_by_truststore(&owners, "my-uid"));
+    }
+
+    #[test]
+    fn different_uid_is_rejected() {
+        let owners = [owner_ref(
+            "secrets.stackable.tech/v1alpha1",
+            "TrustStore",
+            "other-uid",
+            Some(true),
+        )];
+        assert!(!is_owned_by_truststore(&owners, "my-uid"));
+    }
+
+    #[test]
+    fn different_kind_is_rejected() {
+        let owners = [owner_ref("v1", "ConfigMap", "my-uid", Some(true))];
+        assert!(!is_owned_by_truststore(&owners, "my-uid"));
+    }
+
+    #[test]
+    fn foreign_api_group_is_rejected() {
+        let owners = [owner_ref(
+            "evil.example.com/v1",
+            "TrustStore",
+            "my-uid",
+            Some(true),
+        )];
+        assert!(!is_owned_by_truststore(&owners, "my-uid"));
+    }
+
+    #[test]
+    fn kube_root_ca_style_no_owner_is_rejected() {
+        assert!(!is_owned_by_truststore(&[], "my-uid"));
+    }
+}