diff --git a/internal/controllers/machineset/machineset_controller.go b/internal/controllers/machineset/machineset_controller.go
index cbf04cad1ec9..36646a9e2ac1 100644
--- a/internal/controllers/machineset/machineset_controller.go
+++ b/internal/controllers/machineset/machineset_controller.go
@@ -26,6 +26,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apiserver/pkg/storage/names"
@@ -41,6 +42,7 @@ import (
 	"sigs.k8s.io/cluster-api/controllers/external"
 	"sigs.k8s.io/cluster-api/controllers/noderefutil"
 	"sigs.k8s.io/cluster-api/controllers/remote"
+	"sigs.k8s.io/cluster-api/internal/contract"
 	"sigs.k8s.io/cluster-api/internal/controllers/machine"
 	capilabels "sigs.k8s.io/cluster-api/internal/labels"
 	"sigs.k8s.io/cluster-api/internal/util/ssa"
@@ -364,11 +366,14 @@ func (r *Reconciler) reconcile(ctx context.Context, cluster *clusterv1.Cluster,
 	return ctrl.Result{}, nil
 }
 
-// syncMachines updates Machines to propagate in-place mutable fields from the MachineSet.
-// Note: It also cleans up managed fields of all Machines so that Machines that were created/patched before (< v1.4.0)
-// the controller adopted Server-Side-Apply (SSA) can also work with SSA. Otherwise fields would be co-owned by
-// our "old" "manager" and "capi-machineset" and then we would not be able to e.g. drop labels and annotations.
-// TODO: update the labels and annotations to the corresponding infra machines and the boostrap configs of the filtered machines.
+// syncMachines updates Machines, InfrastructureMachine and BootstrapConfig to propagate in-place mutable fields
+// from the MachineSet.
+// Note: It also cleans up managed fields of all Machines so that Machines that were
+// created/patched before (< v1.4.0) the controller adopted Server-Side-Apply (SSA) can also work with SSA.
+// Note: For InfrastructureMachines and BootstrapConfigs it also drops ownership of "metadata.labels" and
+// "metadata.annotations" from "manager" so that "capi-machineset" can own these fields and can work with SSA.
+// Otherwise fields would be co-owned by our "old" "manager" and "capi-machineset" and then we would not be
+// able to e.g. drop labels and annotations.
 func (r *Reconciler) syncMachines(ctx context.Context, machineSet *clusterv1.MachineSet, machines []*clusterv1.Machine) error {
 	log := ctrl.LoggerFrom(ctx)
 	for i := range machines {
@@ -397,6 +402,46 @@ func (r *Reconciler) syncMachines(ctx context.Context, machineSet *clusterv1.Mac
 			return errors.Wrapf(err, "failed to update Machine %q", klog.KObj(updatedMachine))
 		}
 		machines[i] = updatedMachine
+
+		infraMachine, err := external.Get(ctx, r.Client, &updatedMachine.Spec.InfrastructureRef, updatedMachine.Namespace)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get InfrastructureMachine %s",
+				klog.KRef(updatedMachine.Spec.InfrastructureRef.Namespace, updatedMachine.Spec.InfrastructureRef.Name))
+		}
+		// Cleanup managed fields of all InfrastructureMachines to drop ownership of labels and annotations
+		// from "manager". We do this so that InfrastructureMachines that are created using the Create method
+		// can also work with SSA. Otherwise, labels and annotations would be co-owned by our "old" "manager"
+		// and "capi-machineset" and then we would not be able to e.g. drop labels and annotations.
+		labelsAndAnnotationsManagedFieldPaths := []contract.Path{
+			{"f:metadata", "f:annotations"},
+			{"f:metadata", "f:labels"},
+		}
+		if err := ssa.DropManagedFields(ctx, r.Client, infraMachine, machineSetManagerName, labelsAndAnnotationsManagedFieldPaths); err != nil {
+			return errors.Wrapf(err, "failed to update machine: failed to adjust the managedFields of the InfrastructureMachine %s", klog.KObj(infraMachine))
+		}
+		// Update in-place mutating fields on InfrastructureMachine.
+		if err := r.updateExternalObject(ctx, infraMachine, machineSet); err != nil {
+			return errors.Wrapf(err, "failed to update InfrastructureMachine %s", klog.KObj(infraMachine))
+		}
+
+		if updatedMachine.Spec.Bootstrap.ConfigRef != nil {
+			bootstrapConfig, err := external.Get(ctx, r.Client, updatedMachine.Spec.Bootstrap.ConfigRef, updatedMachine.Namespace)
+			if err != nil {
+				return errors.Wrapf(err, "failed to get BootstrapConfig %s",
+					klog.KRef(updatedMachine.Spec.Bootstrap.ConfigRef.Namespace, updatedMachine.Spec.Bootstrap.ConfigRef.Name))
+			}
+			// Cleanup managed fields of all BootstrapConfigs to drop ownership of labels and annotations
+			// from "manager". We do this so that BootstrapConfigs that are created using the Create method
+			// can also work with SSA. Otherwise, labels and annotations would be co-owned by our "old" "manager"
+			// and "capi-machineset" and then we would not be able to e.g. drop labels and annotations.
+			if err := ssa.DropManagedFields(ctx, r.Client, bootstrapConfig, machineSetManagerName, labelsAndAnnotationsManagedFieldPaths); err != nil {
+				return errors.Wrapf(err, "failed to update machine: failed to adjust the managedFields of the BootstrapConfig %s", klog.KObj(bootstrapConfig))
+			}
+			// Update in-place mutating fields on BootstrapConfig.
+			if err := r.updateExternalObject(ctx, bootstrapConfig, machineSet); err != nil {
+				return errors.Wrapf(err, "failed to update BootstrapConfig %s", klog.KObj(bootstrapConfig))
+			}
+		}
 	}
 	return nil
 }
@@ -604,34 +649,72 @@ func (r *Reconciler) computeDesiredMachine(machineSet *clusterv1.MachineSet, exi
 	// When we update an existing Machine will we update the fields on the existing Machine (in-place mutate).
 
 	// Set Labels
+	desiredMachine.Labels = machineLabelsFromMachineSet(machineSet)
+
+	// Set Annotations
+	desiredMachine.Annotations = machineAnnotationsFromMachineSet(machineSet)
+
+	// Set all other in-place mutable fields.
+	desiredMachine.Spec.NodeDrainTimeout = machineSet.Spec.Template.Spec.NodeDrainTimeout
+	desiredMachine.Spec.NodeDeletionTimeout = machineSet.Spec.Template.Spec.NodeDeletionTimeout
+	desiredMachine.Spec.NodeVolumeDetachTimeout = machineSet.Spec.Template.Spec.NodeVolumeDetachTimeout
+
+	return desiredMachine
+}
+
+// updateExternalObject updates the external object passed in with the
+// updated labels and annotations from the MachineSet.
+func (r *Reconciler) updateExternalObject(ctx context.Context, obj client.Object, machineSet *clusterv1.MachineSet) error {
+	updatedObject := &unstructured.Unstructured{}
+	updatedObject.SetGroupVersionKind(obj.GetObjectKind().GroupVersionKind())
+	updatedObject.SetNamespace(obj.GetNamespace())
+	updatedObject.SetName(obj.GetName())
+	// Set the UID to ensure that Server-Side-Apply only performs an update
+	// and does not perform an accidental create.
+	updatedObject.SetUID(obj.GetUID())
+
+	updatedObject.SetLabels(machineLabelsFromMachineSet(machineSet))
+	updatedObject.SetAnnotations(machineAnnotationsFromMachineSet(machineSet))
+
+	patchOptions := []client.PatchOption{
+		client.ForceOwnership,
+		client.FieldOwner(machineSetManagerName),
+	}
+	if err := r.Client.Patch(ctx, updatedObject, client.Apply, patchOptions...); err != nil {
+		return errors.Wrapf(err, "failed to update %s", klog.KObj(obj))
+	}
+	return nil
+}
+
+// machineLabelsFromMachineSet computes the labels the Machine created from this MachineSet should have.
+func machineLabelsFromMachineSet(machineSet *clusterv1.MachineSet) map[string]string {
+	machineLabels := map[string]string{}
 	// Note: We can't just set `machineSet.Spec.Template.Labels` directly and thus "share" the labels
 	// map between Machine and machineSet.Spec.Template.Labels. This would mean that adding the
 	// MachineSetNameLabel and MachineDeploymentNameLabel later on the Machine would also add the labels
 	// to machineSet.Spec.Template.Labels and thus modify the labels of the MachineSet.
 	for k, v := range machineSet.Spec.Template.Labels {
-		desiredMachine.Labels[k] = v
+		machineLabels[k] = v
 	}
 	// Always set the MachineSetNameLabel.
 	// Note: If a client tries to create a MachineSet without a selector, the MachineSet webhook
 	// will add this label automatically. But we want this label to always be present even if the MachineSet
 	// has a selector which doesn't include it. Therefore, we have to set it here explicitly.
-	desiredMachine.Labels[clusterv1.MachineSetNameLabel] = capilabels.MustFormatValue(machineSet.Name)
+	machineLabels[clusterv1.MachineSetNameLabel] = capilabels.MustFormatValue(machineSet.Name)
 	// Propagate the MachineDeploymentNameLabel from MachineSet to Machine if it exists.
 	if mdName, ok := machineSet.Labels[clusterv1.MachineDeploymentNameLabel]; ok {
-		desiredMachine.Labels[clusterv1.MachineDeploymentNameLabel] = mdName
+		machineLabels[clusterv1.MachineDeploymentNameLabel] = mdName
 	}
+	return machineLabels
+}
 
-	// Set Annotations
+// machineAnnotationsFromMachineSet computes the annotations the Machine created from this MachineSet should have.
+func machineAnnotationsFromMachineSet(machineSet *clusterv1.MachineSet) map[string]string {
+	annotations := map[string]string{}
 	for k, v := range machineSet.Spec.Template.Annotations {
-		desiredMachine.Annotations[k] = v
+		annotations[k] = v
 	}
-
-	// Set all other in-place mutable fields.
-	desiredMachine.Spec.NodeDrainTimeout = machineSet.Spec.Template.Spec.NodeDrainTimeout
-	desiredMachine.Spec.NodeDeletionTimeout = machineSet.Spec.Template.Spec.NodeDeletionTimeout
-	desiredMachine.Spec.NodeVolumeDetachTimeout = machineSet.Spec.Template.Spec.NodeVolumeDetachTimeout
-
-	return desiredMachine
+	return annotations
 }
 
 // shouldExcludeMachine returns true if the machine should be filtered out, false otherwise.
diff --git a/internal/controllers/machineset/machineset_controller_test.go b/internal/controllers/machineset/machineset_controller_test.go
index 5ccb76a41ab7..cbcb2fc4dfa6 100644
--- a/internal/controllers/machineset/machineset_controller_test.go
+++ b/internal/controllers/machineset/machineset_controller_test.go
@@ -33,6 +33,7 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/controllers/external"
+	"sigs.k8s.io/cluster-api/internal/contract"
 	"sigs.k8s.io/cluster-api/internal/test/builder"
 	"sigs.k8s.io/cluster-api/internal/util/ssa"
 	"sigs.k8s.io/cluster-api/util"
@@ -983,6 +984,7 @@ func TestMachineSetReconciler_syncMachines(t *testing.T) {
 	namespace, testCluster := setup(t, g)
 	defer teardown(t, g, namespace, testCluster)
 
+	classicManager := "manager"
 	replicas := int32(2)
 	version := "v1.25.3"
 	duration10s := &metav1.Duration{Duration: 10 * time.Second}
@@ -992,7 +994,8 @@ func TestMachineSetReconciler_syncMachines(t *testing.T) {
 			Name:      "ms-1",
 			Namespace: namespace.Name,
 			Labels: map[string]string{
-				"label-1": "true",
+				"label-1":                            "true",
+				clusterv1.MachineDeploymentNameLabel: "md-1",
 			},
 		},
 		Spec: clusterv1.MachineSetSpec{
@@ -1000,17 +1003,20 @@ func TestMachineSetReconciler_syncMachines(t *testing.T) {
 			Replicas:    &replicas,
 			Selector: metav1.LabelSelector{
 				MatchLabels: map[string]string{
-					"machine-set-matching-label": "true",
+					"preserved-label": "preserved-value",
 				},
 			},
 			Template: clusterv1.MachineTemplateSpec{
 				ObjectMeta: clusterv1.ObjectMeta{
 					Labels: map[string]string{
-						"machine-set-matching-label": "true",
+						"preserved-label": "preserved-value", // Label will be preserved while testing in-place mutation.
+						"dropped-label":   "dropped-value",   // Label will be dropped while testing in-place mutation.
+						"modified-label":  "modified-value",  // Label value will be modified while testing in-place mutation.
 					},
 					Annotations: map[string]string{
-						"annotation-1": "true",
-						"precedence":   "MachineSet",
+						"preserved-annotation": "preserved-value", // Annotation will be preserved while testing in-place mutation.
+						"dropped-annotation":   "dropped-value",   // Annotation will be dropped while testing in-place mutation.
+						"modified-annotation":  "modified-value",  // Annotation value will be modified while testing in-place mutation.
 					},
 				},
 				Spec: clusterv1.MachineSpec{
@@ -1028,14 +1034,63 @@ func TestMachineSetReconciler_syncMachines(t *testing.T) {
 						Kind:       "GenericInfrastructureMachineTemplate",
 						Name:       "ms-template",
 					},
-					NodeDrainTimeout:        duration10s,
-					NodeDeletionTimeout:     duration10s,
-					NodeVolumeDetachTimeout: duration10s,
 				},
 			},
 		},
 	}
 
+	infraMachineSpec := map[string]interface{}{
+		"infra-field": "infra-value",
+	}
+	infraMachine := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"kind":       "GenericInfrastructureMachine",
+			"apiVersion": "infrastructure.cluster.x-k8s.io/v1beta1",
+			"metadata": map[string]interface{}{
+				"name":      "infra-machine-1",
+				"namespace": namespace.Name,
+				"labels": map[string]string{
+					"preserved-label": "preserved-value",
+					"dropped-label":   "dropped-value",
+					"modified-label":  "modified-value",
+				},
+				"annotations": map[string]string{
+					"preserved-annotation": "preserved-value",
+					"dropped-annotation":   "dropped-value",
+					"modified-annotation":  "modified-value",
+				},
+			},
+			"spec": infraMachineSpec,
+		},
+	}
+	g.Expect(env.Create(ctx, infraMachine, client.FieldOwner(classicManager))).To(Succeed())
+
+	bootstrapConfigSpec := map[string]interface{}{
+		"bootstrap-field": "bootstrap-value",
+	}
+	bootstrapConfig := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"kind":       "GenericBootstrapConfig",
+			"apiVersion": "bootstrap.cluster.x-k8s.io/v1beta1",
+			"metadata": map[string]interface{}{
+				"name":      "bootstrap-config-1",
+				"namespace": namespace.Name,
+				"labels": map[string]string{
+					"preserved-label": "preserved-value",
+					"dropped-label":   "dropped-value",
+					"modified-label":  "modified-value",
+				},
+				"annotations": map[string]string{
+					"preserved-annotation": "preserved-value",
+					"dropped-annotation":   "dropped-value",
+					"modified-annotation":  "modified-value",
+				},
+			},
+			"spec": bootstrapConfigSpec,
+		},
+	}
+	g.Expect(env.Create(ctx, bootstrapConfig, client.FieldOwner(classicManager))).To(Succeed())
+
 	inPlaceMutatingMachine := &clusterv1.Machine{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: clusterv1.GroupVersion.String(),
@@ -1045,23 +1100,38 @@ func TestMachineSetReconciler_syncMachines(t *testing.T) {
 			UID:       "abc-123-uid",
 			Name:      "in-place-mutating-machine",
 			Namespace: namespace.Name,
-			Labels:    map[string]string{},
+			Labels: map[string]string{
+				"preserved-label": "preserved-value",
+				"dropped-label":   "dropped-value",
+				"modified-label":  "modified-value",
+			},
 			Annotations: map[string]string{
-				// This will be overwritten with the annotation from the MachineSet
-				"precedence": "Machine",
+				"preserved-annotation": "preserved-value",
+				"dropped-annotation":   "dropped-value",
+				"modified-annotation":  "modified-value",
 			},
 		},
 		Spec: clusterv1.MachineSpec{
 			ClusterName: testClusterName,
 			InfrastructureRef: corev1.ObjectReference{
-				Namespace: namespace.Name,
+				Namespace:  infraMachine.GetNamespace(),
+				Name:       infraMachine.GetName(),
+				UID:        infraMachine.GetUID(),
+				APIVersion: infraMachine.GetAPIVersion(),
+				Kind:       infraMachine.GetKind(),
 			},
 			Bootstrap: clusterv1.Bootstrap{
-				DataSecretName: pointer.String("machine-bootstrap-secret"),
+				ConfigRef: &corev1.ObjectReference{
+					Namespace:  bootstrapConfig.GetNamespace(),
+					Name:       bootstrapConfig.GetName(),
+					UID:        bootstrapConfig.GetUID(),
+					APIVersion: bootstrapConfig.GetAPIVersion(),
+					Kind:       bootstrapConfig.GetKind(),
+				},
 			},
 		},
 	}
-	g.Expect(env.Create(ctx, inPlaceMutatingMachine, client.FieldOwner("manager"))).To(Succeed())
+	g.Expect(env.Create(ctx, inPlaceMutatingMachine, client.FieldOwner(classicManager))).To(Succeed())
 
 	deletingMachine := &clusterv1.Machine{
 		TypeMeta: metav1.TypeMeta{
@@ -1086,7 +1156,7 @@ func TestMachineSetReconciler_syncMachines(t *testing.T) {
 			},
 		},
 	}
-	g.Expect(env.Create(ctx, deletingMachine, client.FieldOwner("manager"))).To(Succeed())
+	g.Expect(env.Create(ctx, deletingMachine, client.FieldOwner(classicManager))).To(Succeed())
 	// Delete the machine to put it in the deleting state
 	g.Expect(env.Delete(ctx, deletingMachine)).To(Succeed())
 	// Wait till the machine is marked for deletion
@@ -1099,45 +1169,135 @@ func TestMachineSetReconciler_syncMachines(t *testing.T) {
 
 	machines := []*clusterv1.Machine{inPlaceMutatingMachine, deletingMachine}
 
-	// Sync the Machines.
+	//
+	// Verify Managed Fields
+	//
+
+	// Run syncMachines to clean up managed fields and have proper field ownership
+	// for Machines, InfrastructureMachines and BootstrapConfigs.
 	reconciler := &Reconciler{Client: env}
 	g.Expect(reconciler.syncMachines(ctx, ms, machines)).To(Succeed())
 
-	// The in-place mutating machine should have:
-	// - clean-up managed fields
-	// - updated in-place propagating values
+	// The inPlaceMutatingMachine should have cleaned up managed fields.
 	updatedInPlaceMutatingMachine := inPlaceMutatingMachine.DeepCopy()
 	g.Expect(env.GetAPIReader().Get(ctx, client.ObjectKeyFromObject(updatedInPlaceMutatingMachine), updatedInPlaceMutatingMachine))
-
 	// Verify ManagedFields
 	g.Expect(updatedInPlaceMutatingMachine.ManagedFields).Should(
 		ContainElement(ssa.MatchManagedFieldsEntry(machineSetManagerName, metav1.ManagedFieldsOperationApply)),
 		"in-place mutable machine should contain an entry for SSA manager",
 	)
 	g.Expect(updatedInPlaceMutatingMachine.ManagedFields).ShouldNot(
-		ContainElement(ssa.MatchManagedFieldsEntry("manager", metav1.ManagedFieldsOperationUpdate)),
+		ContainElement(ssa.MatchManagedFieldsEntry(classicManager, metav1.ManagedFieldsOperationUpdate)),
 		"in-place mutable machine should not contain an entry for old manager",
 	)
 
-	// Verify in-place mutable fields have been updated.
-	// Verify Labels
-	g.Expect(updatedInPlaceMutatingMachine.Labels).Should(HaveKeyWithValue("machine-set-matching-label", "true"))
-	// Verify Annotations
-	g.Expect(updatedInPlaceMutatingMachine.Annotations).Should(HaveKeyWithValue("annotation-1", "true"))
-	g.Expect(updatedInPlaceMutatingMachine.Annotations).Should(HaveKeyWithValue("precedence", "MachineSet"))
-	// Verify Node timeout values
-	g.Expect(updatedInPlaceMutatingMachine.Spec.NodeDrainTimeout).Should(And(
-		Not(BeNil()),
-		HaveValue(Equal(*ms.Spec.Template.Spec.NodeDrainTimeout)),
-	))
-	g.Expect(updatedInPlaceMutatingMachine.Spec.NodeDeletionTimeout).Should(And(
-		Not(BeNil()),
-		HaveValue(Equal(*ms.Spec.Template.Spec.NodeDeletionTimeout)),
-	))
-	g.Expect(updatedInPlaceMutatingMachine.Spec.NodeVolumeDetachTimeout).Should(And(
-		Not(BeNil()),
-		HaveValue(Equal(*ms.Spec.Template.Spec.NodeDrainTimeout)),
-	))
+	// The InfrastructureMachine should have ownership of "labels" and "annotations" transferred to
+	// "capi-machineset" manager.
+	updatedInfraMachine := infraMachine.DeepCopy()
+	g.Expect(env.GetAPIReader().Get(ctx, client.ObjectKeyFromObject(updatedInfraMachine), updatedInfraMachine))
+
+	// Verify ManagedFields
+	g.Expect(updatedInfraMachine.GetManagedFields()).Should(
+		ssa.MatchFieldOwnership(machineSetManagerName, metav1.ManagedFieldsOperationApply, contract.Path{"f:metadata", "f:labels"}))
+	g.Expect(updatedInfraMachine.GetManagedFields()).Should(
+		ssa.MatchFieldOwnership(machineSetManagerName, metav1.ManagedFieldsOperationApply, contract.Path{"f:metadata", "f:annotations"}))
+	g.Expect(updatedInfraMachine.GetManagedFields()).ShouldNot(
+		ssa.MatchFieldOwnership(classicManager, metav1.ManagedFieldsOperationUpdate, contract.Path{"f:metadata", "f:labels"}))
+	g.Expect(updatedInfraMachine.GetManagedFields()).ShouldNot(
+		ssa.MatchFieldOwnership(classicManager, metav1.ManagedFieldsOperationUpdate, contract.Path{"f:metadata", "f:annotations"}))
+	g.Expect(updatedInfraMachine.GetManagedFields()).Should(
+		ssa.MatchFieldOwnership(classicManager, metav1.ManagedFieldsOperationUpdate, contract.Path{"f:spec"}))
+
+	// The BootstrapConfig should have ownership of "labels" and "annotations" transferred to
+	// "capi-machineset" manager.
+	updatedBootstrapConfig := bootstrapConfig.DeepCopy()
+	g.Expect(env.GetAPIReader().Get(ctx, client.ObjectKeyFromObject(updatedBootstrapConfig), updatedBootstrapConfig))
+
+	// Verify ManagedFields
+	g.Expect(updatedBootstrapConfig.GetManagedFields()).Should(
+		ssa.MatchFieldOwnership(machineSetManagerName, metav1.ManagedFieldsOperationApply, contract.Path{"f:metadata", "f:labels"}))
+	g.Expect(updatedBootstrapConfig.GetManagedFields()).Should(
+		ssa.MatchFieldOwnership(machineSetManagerName, metav1.ManagedFieldsOperationApply, contract.Path{"f:metadata", "f:annotations"}))
+	g.Expect(updatedBootstrapConfig.GetManagedFields()).ShouldNot(
+		ssa.MatchFieldOwnership(classicManager, metav1.ManagedFieldsOperationUpdate, contract.Path{"f:metadata", "f:labels"}))
+	g.Expect(updatedBootstrapConfig.GetManagedFields()).ShouldNot(
+		ssa.MatchFieldOwnership(classicManager, metav1.ManagedFieldsOperationUpdate, contract.Path{"f:metadata", "f:annotations"}))
+	g.Expect(updatedBootstrapConfig.GetManagedFields()).Should(
+		ssa.MatchFieldOwnership(classicManager, metav1.ManagedFieldsOperationUpdate, contract.Path{"f:spec"}))
+
+	//
+	// Verify In-place mutating fields
+	//
+
+	// Update the MachineSet and verify the in-mutating fields are propagated.
+	ms.Spec.Template.Labels = map[string]string{
+		"preserved-label": "preserved-value",  // Keep the label and value as is
+		"modified-label":  "modified-value-2", // Modify the value of the label
+		// Drop "dropped-label"
+	}
+	expectedLabels := map[string]string{
+		"preserved-label":                    "preserved-value",
+		"modified-label":                     "modified-value-2",
+		clusterv1.MachineSetNameLabel:        ms.Name,
+		clusterv1.MachineDeploymentNameLabel: "md-1",
+		clusterv1.ClusterNameLabel:           testClusterName, // This label is added by the Machine controller.
+	}
+	ms.Spec.Template.Annotations = map[string]string{
+		"preserved-annotation": "preserved-value",  // Keep the annotation and value as is
+		"modified-annotation":  "modified-value-2", // Modify the value of the annotation
+		// Drop "dropped-annotation"
+	}
+	ms.Spec.Template.Spec.NodeDrainTimeout = duration10s
+	ms.Spec.Template.Spec.NodeDeletionTimeout = duration10s
+	ms.Spec.Template.Spec.NodeVolumeDetachTimeout = duration10s
+	g.Expect(reconciler.syncMachines(ctx, ms, []*clusterv1.Machine{updatedInPlaceMutatingMachine, deletingMachine})).To(Succeed())
+
+	// Verify in-place mutable fields are updated on the Machine.
+	updatedInPlaceMutatingMachine = inPlaceMutatingMachine.DeepCopy()
+	g.Eventually(func(g Gomega) {
+		g.Expect(env.GetAPIReader().Get(ctx, client.ObjectKeyFromObject(updatedInPlaceMutatingMachine), updatedInPlaceMutatingMachine))
+		// Verify Labels
+		g.Expect(updatedInPlaceMutatingMachine.Labels).Should(Equal(expectedLabels))
+		// Verify Annotations
+		g.Expect(updatedInPlaceMutatingMachine.Annotations).Should(Equal(ms.Spec.Template.Annotations))
+		// Verify Node timeout values
+		g.Expect(updatedInPlaceMutatingMachine.Spec.NodeDrainTimeout).Should(And(
+			Not(BeNil()),
+			HaveValue(Equal(*ms.Spec.Template.Spec.NodeDrainTimeout)),
+		))
+		g.Expect(updatedInPlaceMutatingMachine.Spec.NodeDeletionTimeout).Should(And(
+			Not(BeNil()),
+			HaveValue(Equal(*ms.Spec.Template.Spec.NodeDeletionTimeout)),
+		))
+		g.Expect(updatedInPlaceMutatingMachine.Spec.NodeVolumeDetachTimeout).Should(And(
+			Not(BeNil()),
+			HaveValue(Equal(*ms.Spec.Template.Spec.NodeDrainTimeout)),
+		))
+	}, timeout).Should(Succeed())
+
+	// Verify in-place mutable fields are updated on InfrastructureMachine
+	updatedInfraMachine = infraMachine.DeepCopy()
+	g.Eventually(func(g Gomega) {
+		g.Expect(env.GetAPIReader().Get(ctx, client.ObjectKeyFromObject(updatedInfraMachine), updatedInfraMachine))
+		// Verify Labels
+		g.Expect(updatedInfraMachine.GetLabels()).Should(Equal(expectedLabels))
+		// Verify Annotations
+		g.Expect(updatedInfraMachine.GetAnnotations()).Should(Equal(ms.Spec.Template.Annotations))
+		// Verify spec remains the same
+		g.Expect(updatedInfraMachine.Object).Should(HaveKeyWithValue("spec", infraMachineSpec))
+	}, timeout).Should(Succeed())
+
+	// Verify in-place mutable fields are updated on the BootstrapConfig.
+	updatedBootstrapConfig = bootstrapConfig.DeepCopy()
+	g.Eventually(func(g Gomega) {
+		g.Expect(env.GetAPIReader().Get(ctx, client.ObjectKeyFromObject(updatedBootstrapConfig), updatedBootstrapConfig))
+		// Verify Labels
+		g.Expect(updatedBootstrapConfig.GetLabels()).Should(Equal(expectedLabels))
+		// Verify Annotations
+		g.Expect(updatedBootstrapConfig.GetAnnotations()).Should(Equal(ms.Spec.Template.Annotations))
+		// Verify spec remains the same
+		g.Expect(updatedBootstrapConfig.Object).Should(HaveKeyWithValue("spec", bootstrapConfigSpec))
+	}, timeout).Should(Succeed())
 
 	// Wait to ensure Machine is not updated.
 	// Verify that the machine stays the same consistently.
diff --git a/internal/controllers/topology/cluster/structuredmerge/drop_diff_test.go b/internal/controllers/topology/cluster/structuredmerge/drop_diff_test.go
index 2f74c5eac82e..a4fbbbe893ea 100644
--- a/internal/controllers/topology/cluster/structuredmerge/drop_diff_test.go
+++ b/internal/controllers/topology/cluster/structuredmerge/drop_diff_test.go
@@ -22,6 +22,7 @@ import (
 	. "github.com/onsi/gomega"
 
 	"sigs.k8s.io/cluster-api/internal/contract"
+	"sigs.k8s.io/cluster-api/internal/util/ssa"
 )
 
 func Test_dropDiffForNotAllowedPaths(t *testing.T) {
@@ -59,7 +60,7 @@ func Test_dropDiffForNotAllowedPaths(t *testing.T) {
 						"foo": "123-changed",
 					},
 				},
-				shouldDropDiffFunc: isNotAllowedPath(
+				shouldDropDiffFunc: ssa.IsNotAllowedPath(
 					[]contract.Path{ // NOTE: we are dropping everything not in this list (IsNotAllowed)
 						{"metadata", "labels"},
 						{"metadata", "annotations"},
@@ -109,7 +110,7 @@ func Test_dropDiffForNotAllowedPaths(t *testing.T) {
 						"foo": "123",
 					},
 				},
-				shouldDropDiffFunc: isNotAllowedPath(
+				shouldDropDiffFunc: ssa.IsNotAllowedPath(
 					[]contract.Path{ // NOTE: we are dropping everything not in this list (IsNotAllowed)
 						{"metadata", "labels"},
 						{"metadata", "annotations"},
@@ -145,7 +146,7 @@ func Test_dropDiffForNotAllowedPaths(t *testing.T) {
 						"foo": "123",
 					},
 				},
-				shouldDropDiffFunc: isNotAllowedPath(
+				shouldDropDiffFunc: ssa.IsNotAllowedPath(
 					[]contract.Path{}, // NOTE: we are dropping everything not in this list (IsNotAllowed)
 				),
 			},
@@ -193,7 +194,7 @@ func Test_dropDiffForIgnoredPaths(t *testing.T) {
 						},
 					},
 				},
-				shouldDropDiffFunc: isIgnorePath(
+				shouldDropDiffFunc: ssa.IsIgnorePath(
 					[]contract.Path{
 						{"spec", "controlPlaneEndpoint"},
 					},
@@ -225,7 +226,7 @@ func Test_dropDiffForIgnoredPaths(t *testing.T) {
 						},
 					},
 				},
-				shouldDropDiffFunc: isIgnorePath(
+				shouldDropDiffFunc: ssa.IsIgnorePath(
 					[]contract.Path{
 						{"spec", "controlPlaneEndpoint"},
 					},
@@ -250,7 +251,7 @@ func Test_dropDiffForIgnoredPaths(t *testing.T) {
 						"foo": "123",
 					},
 				},
-				shouldDropDiffFunc: isIgnorePath(
+				shouldDropDiffFunc: ssa.IsIgnorePath(
 					[]contract.Path{
 						{"spec", "foo"},
 					},
diff --git a/internal/controllers/topology/cluster/structuredmerge/dryrun.go b/internal/controllers/topology/cluster/structuredmerge/dryrun.go
index 08a3f6bbb34c..d7c183c362dc 100644
--- a/internal/controllers/topology/cluster/structuredmerge/dryrun.go
+++ b/internal/controllers/topology/cluster/structuredmerge/dryrun.go
@@ -28,6 +28,7 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/internal/contract"
+	"sigs.k8s.io/cluster-api/internal/util/ssa"
 	"sigs.k8s.io/cluster-api/util/conversion"
 )
 
@@ -47,9 +48,9 @@ type dryRunSSAPatchInput struct {
 func dryRunSSAPatch(ctx context.Context, dryRunCtx *dryRunSSAPatchInput) (bool, bool, error) {
 	// For dry run we use the same options as for the intent but with adding metadata.managedFields
 	// to ensure that changes to ownership are detected.
-	dryRunHelperOptions := &HelperOptions{
-		allowedPaths: append(dryRunCtx.helperOptions.allowedPaths, []string{"metadata", "managedFields"}),
-		ignorePaths:  dryRunCtx.helperOptions.ignorePaths,
+	filterObjectInput := &ssa.FilterObjectInput{
+		AllowedPaths: append(dryRunCtx.helperOptions.allowedPaths, []string{"metadata", "managedFields"}),
+		IgnorePaths:  dryRunCtx.helperOptions.ignorePaths,
 	}
 
 	// Add TopologyDryRunAnnotation to notify validation webhooks to skip immutability checks.
@@ -80,7 +81,7 @@ func dryRunSSAPatch(ctx context.Context, dryRunCtx *dryRunSSAPatchInput) (bool,
 	// Filter object to drop fields which are not part of our intent.
 	// Note: It's especially important to also drop metadata.resourceVersion, otherwise we could get the following
 	// error: "the object has been modified; please apply your changes to the latest version and try again"
-	filterObject(dryRunCtx.originalUnstructured, dryRunHelperOptions)
+	ssa.FilterObject(dryRunCtx.originalUnstructured, filterObjectInput)
 	// Backup managed fields.
 	originalUnstructuredManagedFieldsBeforeSSA := dryRunCtx.originalUnstructured.GetManagedFields()
 	// Set managed fields to nil.
@@ -119,8 +120,8 @@ func dryRunSSAPatch(ctx context.Context, dryRunCtx *dryRunSSAPatchInput) (bool,
 	}
 
 	// Drop the other fields which are not part of our intent.
-	filterObject(dryRunCtx.modifiedUnstructured, dryRunHelperOptions)
-	filterObject(dryRunCtx.originalUnstructured, dryRunHelperOptions)
+	ssa.FilterObject(dryRunCtx.modifiedUnstructured, filterObjectInput)
+	ssa.FilterObject(dryRunCtx.originalUnstructured, filterObjectInput)
 
 	// Compare the output of dry run to the original object.
 	originalJSON, err := json.Marshal(dryRunCtx.originalUnstructured)
@@ -155,10 +156,10 @@ func dryRunSSAPatch(ctx context.Context, dryRunCtx *dryRunSSAPatchInput) (bool,
 // it is expected to change due to the additional annotation.
 func cleanupManagedFieldsAndAnnotation(obj *unstructured.Unstructured) error {
 	// Filter the topology.cluster.x-k8s.io/dry-run annotation as well as leftover empty maps.
-	filterIntent(&filterIntentInput{
-		path:  contract.Path{},
-		value: obj.Object,
-		shouldFilter: isIgnorePath([]contract.Path{
+	ssa.FilterIntent(&ssa.FilterIntentInput{
+		Path:  contract.Path{},
+		Value: obj.Object,
+		ShouldFilter: ssa.IsIgnorePath([]contract.Path{
 			{"metadata", "annotations", clusterv1.TopologyDryRunAnnotation},
 			// In case the ClusterClass we are reconciling is using not the latest apiVersion the conversion
 			// annotation might be added to objects. As we don't care about differences in conversion as we
@@ -192,10 +193,10 @@ func cleanupManagedFieldsAndAnnotation(obj *unstructured.Unstructured) error {
 		}
 
 		// Filter out the annotation ownership as well as leftover empty maps.
-		filterIntent(&filterIntentInput{
-			path:  contract.Path{},
-			value: fieldsV1,
-			shouldFilter: isIgnorePath([]contract.Path{
+		ssa.FilterIntent(&ssa.FilterIntentInput{
+			Path:  contract.Path{},
+			Value: fieldsV1,
+			ShouldFilter: ssa.IsIgnorePath([]contract.Path{
 				{"f:metadata", "f:annotations", "f:" + clusterv1.TopologyDryRunAnnotation},
 				{"f:metadata", "f:annotations", "f:" + conversion.DataAnnotation},
 			}),
diff --git a/internal/controllers/topology/cluster/structuredmerge/filterintent.go b/internal/controllers/topology/cluster/structuredmerge/filterintent.go
deleted file mode 100644
index a8d741d1e6a1..000000000000
--- a/internal/controllers/topology/cluster/structuredmerge/filterintent.go
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
-Copyright 2022 The Kubernetes Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package structuredmerge
-
-import (
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-
-	"sigs.k8s.io/cluster-api/internal/contract"
-)
-
-// filterObject filter out changes not relevant for the topology controller.
-func filterObject(obj *unstructured.Unstructured, helperOptions *HelperOptions) {
-	// filter out changes not in the allowed paths (fields to not consider, e.g. status);
-	if len(helperOptions.allowedPaths) > 0 {
-		filterIntent(&filterIntentInput{
-			path:         contract.Path{},
-			value:        obj.Object,
-			shouldFilter: isNotAllowedPath(helperOptions.allowedPaths),
-		})
-	}
-
-	// filter out changes for ignore paths (well known fields owned by other controllers, e.g.
-	//   spec.controlPlaneEndpoint in the InfrastructureCluster object);
-	if len(helperOptions.ignorePaths) > 0 {
-		filterIntent(&filterIntentInput{
-			path:         contract.Path{},
-			value:        obj.Object,
-			shouldFilter: isIgnorePath(helperOptions.ignorePaths),
-		})
-	}
-}
-
-// filterIntent ensures that object only includes the fields and values for which the topology controller has an opinion,
-// and filter out everything else by removing it from the unstructured object.
-// NOTE: This func is called recursively only for fields of type Map, but this is ok given the current use cases
-// this func has to address. More specifically, we are using this func for filtering out not allowed paths and for ignore paths;
-// all of them are defined in reconcile_state.go and are targeting well-known fields inside nested maps.
-// Allowed paths / ignore paths which point to an array are not supported by the current implementation.
-func filterIntent(ctx *filterIntentInput) bool {
-	value, ok := ctx.value.(map[string]interface{})
-	if !ok {
-		return false
-	}
-
-	gotDeletions := false
-	for field := range value {
-		fieldCtx := &filterIntentInput{
-			// Compose the path for the nested field.
-			path: ctx.path.Append(field),
-			// Gets the original and the modified value for the field.
-			value: value[field],
-			// Carry over global values from the context.
-			shouldFilter: ctx.shouldFilter,
-		}
-
-		// If the field should be filtered out, delete it from the modified object.
-		if fieldCtx.shouldFilter(fieldCtx.path) {
-			delete(value, field)
-			gotDeletions = true
-			continue
-		}
-
-		// Process nested fields and get in return if filterIntent removed fields.
-		if filterIntent(fieldCtx) {
-			// Ensure we are not leaving empty maps around.
-			if v, ok := fieldCtx.value.(map[string]interface{}); ok && len(v) == 0 {
-				delete(value, field)
-				gotDeletions = true
-			}
-		}
-	}
-	return gotDeletions
-}
-
-// filterIntentInput holds info required while filtering the intent for server side apply.
-// NOTE: in server side apply an intent is a partial object that only includes the fields and values for which the user has an opinion.
-type filterIntentInput struct {
-	// the path of the field being processed.
-	path contract.Path
-
-	// the value for the current path.
-	value interface{}
-
-	// shouldFilter handle the func that determine if the current path should be dropped or not.
-	shouldFilter func(path contract.Path) bool
-}
-
-// isAllowedPath returns true when the path is one of the allowedPaths.
-func isAllowedPath(allowedPaths []contract.Path) func(path contract.Path) bool {
-	return func(path contract.Path) bool {
-		for _, p := range allowedPaths {
-			// NOTE: we allow everything Equal or one IsParentOf one of the allowed paths.
-			// e.g. if allowed path is metadata.labels, we allow both metadata and metadata.labels;
-			// this is required because allowed path is called recursively.
-			if path.Overlaps(p) {
-				return true
-			}
-		}
-		return false
-	}
-}
-
-// isNotAllowedPath returns true when the path is NOT one of the allowedPaths.
-func isNotAllowedPath(allowedPaths []contract.Path) func(path contract.Path) bool {
-	return func(path contract.Path) bool {
-		isAllowed := isAllowedPath(allowedPaths)
-		return !isAllowed(path)
-	}
-}
-
-// isIgnorePath returns true when the path is one of the ignorePaths.
-func isIgnorePath(ignorePaths []contract.Path) func(path contract.Path) bool {
-	return func(path contract.Path) bool {
-		for _, p := range ignorePaths {
-			if path.Equal(p) {
-				return true
-			}
-		}
-		return false
-	}
-}
diff --git a/internal/controllers/topology/cluster/structuredmerge/serversidepathhelper.go b/internal/controllers/topology/cluster/structuredmerge/serversidepathhelper.go
index 679f719ec67a..7d1a2b617cf9 100644
--- a/internal/controllers/topology/cluster/structuredmerge/serversidepathhelper.go
+++ b/internal/controllers/topology/cluster/structuredmerge/serversidepathhelper.go
@@ -24,6 +24,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"sigs.k8s.io/cluster-api/internal/util/ssa"
 	"sigs.k8s.io/cluster-api/util"
 )
 
@@ -70,7 +71,10 @@ func NewServerSidePatchHelper(ctx context.Context, original, modified client.Obj
 
 	// Filter the modifiedUnstructured object to only contain changes intendet to be done.
 	// The originalUnstructured object will be filtered in dryRunSSAPatch using other options.
-	filterObject(modifiedUnstructured, helperOptions)
+	ssa.FilterObject(modifiedUnstructured, &ssa.FilterObjectInput{
+		AllowedPaths: helperOptions.allowedPaths,
+		IgnorePaths:  helperOptions.ignorePaths,
+	})
 
 	// Carry over uid to match the intent to:
 	// * create (uid==""):
diff --git a/internal/controllers/topology/cluster/structuredmerge/twowayspatchhelper.go b/internal/controllers/topology/cluster/structuredmerge/twowayspatchhelper.go
index abb93795220a..dc71444c7ab7 100644
--- a/internal/controllers/topology/cluster/structuredmerge/twowayspatchhelper.go
+++ b/internal/controllers/topology/cluster/structuredmerge/twowayspatchhelper.go
@@ -29,6 +29,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"sigs.k8s.io/cluster-api/internal/contract"
+	"sigs.k8s.io/cluster-api/internal/util/ssa"
 	"sigs.k8s.io/cluster-api/util"
 )
 
@@ -170,7 +171,7 @@ func applyOptions(in *applyOptionsInput) ([]byte, error) {
 			path:               contract.Path{},
 			original:           originalMap,
 			modified:           modifiedMap,
-			shouldDropDiffFunc: isNotAllowedPath(in.options.allowedPaths),
+			shouldDropDiffFunc: ssa.IsNotAllowedPath(in.options.allowedPaths),
 		})
 	}
 
@@ -182,7 +183,7 @@ func applyOptions(in *applyOptionsInput) ([]byte, error) {
 			path:               contract.Path{},
 			original:           originalMap,
 			modified:           modifiedMap,
-			shouldDropDiffFunc: isIgnorePath(in.options.ignorePaths),
+			shouldDropDiffFunc: ssa.IsIgnorePath(in.options.ignorePaths),
 		})
 	}
 
diff --git a/internal/util/ssa/filterintent.go b/internal/util/ssa/filterintent.go
new file mode 100644
index 000000000000..26cb4b15814b
--- /dev/null
+++ b/internal/util/ssa/filterintent.go
@@ -0,0 +1,147 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package ssa
+
+import (
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+
+	"sigs.k8s.io/cluster-api/internal/contract"
+)
+
+// FilterObjectInput holds info required while filtering the object.
+type FilterObjectInput struct {
+	// AllowedPaths instruct FilterObject to ignore everything except given paths.
+	AllowedPaths []contract.Path
+
+	// IgnorePaths instruct FilterObject to ignore given paths.
+	// NOTE: IgnorePaths are used to filter out fields nested inside AllowedPaths, e.g.
+	// spec.ControlPlaneEndpoint.
+	// NOTE: ignore paths which point to an array are not supported by the current implementation.
+	IgnorePaths []contract.Path
+}
+
+// FilterObject filter out changes not relevant for the controller.
+func FilterObject(obj *unstructured.Unstructured, input *FilterObjectInput) {
+	// filter out changes not in the allowed paths (fields to not consider, e.g. status);
+	if len(input.AllowedPaths) > 0 {
+		FilterIntent(&FilterIntentInput{
+			Path:         contract.Path{},
+			Value:        obj.Object,
+			ShouldFilter: IsNotAllowedPath(input.AllowedPaths),
+		})
+	}
+
+	// filter out changes for ignore paths (well known fields owned by other controllers, e.g.
+	//   spec.controlPlaneEndpoint in the InfrastructureCluster object);
+	if len(input.IgnorePaths) > 0 {
+		FilterIntent(&FilterIntentInput{
+			Path:         contract.Path{},
+			Value:        obj.Object,
+			ShouldFilter: IsIgnorePath(input.IgnorePaths),
+		})
+	}
+}
+
+// FilterIntent ensures that object only includes the fields and values for which the controller has an opinion,
+// and filter out everything else by removing it from the Value.
+// NOTE: This func is called recursively only for fields of type Map, but this is ok given the current use cases
+// this func has to address. More specifically, we are using this func for filtering out not allowed paths and for ignore paths;
+// all of them are defined in reconcile_state.go and are targeting well-known fields inside nested maps.
+// Allowed paths / ignore paths which point to an array are not supported by the current implementation.
+func FilterIntent(ctx *FilterIntentInput) bool {
+	value, ok := ctx.Value.(map[string]interface{})
+	if !ok {
+		return false
+	}
+
+	gotDeletions := false
+	for field := range value {
+		fieldCtx := &FilterIntentInput{
+			// Compose the Path for the nested field.
+			Path: ctx.Path.Append(field),
+			// Gets the original and the modified Value for the field.
+			Value: value[field],
+			// Carry over global values from the context.
+			ShouldFilter: ctx.ShouldFilter,
+		}
+
+		// If the field should be filtered out, delete it from the modified object.
+		if fieldCtx.ShouldFilter(fieldCtx.Path) {
+			delete(value, field)
+			gotDeletions = true
+			continue
+		}
+
+		// Process nested fields and get in return if FilterIntent removed fields.
+		if FilterIntent(fieldCtx) {
+			// Ensure we are not leaving empty maps around.
+			if v, ok := fieldCtx.Value.(map[string]interface{}); ok && len(v) == 0 {
+				delete(value, field)
+				gotDeletions = true
+			}
+		}
+	}
+	return gotDeletions
+}
+
+// FilterIntentInput holds info required while filtering the intent for server side apply.
+// NOTE: in server side apply an intent is a partial object that only includes the fields and values for which the user has an opinion.
+type FilterIntentInput struct {
+	// the Path of the field being processed.
+	Path contract.Path
+
+	// the Value for the current Path.
+	Value interface{}
+
+	// ShouldFilter handle the func that determine if the current Path should be dropped or not.
+	ShouldFilter func(path contract.Path) bool
+}
+
+// IsAllowedPath returns true when the Path is one of the AllowedPaths.
+func IsAllowedPath(allowedPaths []contract.Path) func(path contract.Path) bool {
+	return func(path contract.Path) bool {
+		for _, p := range allowedPaths {
+			// NOTE: we allow everything Equal or one IsParentOf one of the allowed paths.
+			// e.g. if allowed Path is metadata.labels, we allow both metadata and metadata.labels;
+			// this is required because allowed Path is called recursively.
+			if path.Overlaps(p) {
+				return true
+			}
+		}
+		return false
+	}
+}
+
+// IsNotAllowedPath returns true when the Path is NOT one of the AllowedPaths.
+func IsNotAllowedPath(allowedPaths []contract.Path) func(path contract.Path) bool {
+	return func(path contract.Path) bool {
+		isAllowed := IsAllowedPath(allowedPaths)
+		return !isAllowed(path)
+	}
+}
+
+// IsIgnorePath returns true when the Path is one of the IgnorePaths.
+func IsIgnorePath(ignorePaths []contract.Path) func(path contract.Path) bool {
+	return func(path contract.Path) bool {
+		for _, p := range ignorePaths {
+			if path.Equal(p) {
+				return true
+			}
+		}
+		return false
+	}
+}
diff --git a/internal/controllers/topology/cluster/structuredmerge/filterintent_test.go b/internal/util/ssa/filterintent_test.go
similarity index 82%
rename from internal/controllers/topology/cluster/structuredmerge/filterintent_test.go
rename to internal/util/ssa/filterintent_test.go
index e68f5bc6335d..80949d43bbee 100644
--- a/internal/controllers/topology/cluster/structuredmerge/filterintent_test.go
+++ b/internal/util/ssa/filterintent_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package structuredmerge
+package ssa
 
 import (
 	"testing"
@@ -27,14 +27,14 @@ import (
 func Test_filterNotAllowedPaths(t *testing.T) {
 	tests := []struct {
 		name      string
-		ctx       *filterIntentInput
+		ctx       *FilterIntentInput
 		wantValue map[string]interface{}
 	}{
 		{
 			name: "Filters out not allowed paths",
-			ctx: &filterIntentInput{
-				path: contract.Path{},
-				value: map[string]interface{}{
+			ctx: &FilterIntentInput{
+				Path: contract.Path{},
+				Value: map[string]interface{}{
 					"apiVersion": "foo.bar/v1",
 					"kind":       "Foo",
 					"metadata": map[string]interface{}{
@@ -55,7 +55,7 @@ func Test_filterNotAllowedPaths(t *testing.T) {
 						"foo": "123",
 					},
 				},
-				shouldFilter: isNotAllowedPath(
+				ShouldFilter: IsNotAllowedPath(
 					[]contract.Path{ // NOTE: we are dropping everything not in this list
 						{"apiVersion"},
 						{"kind"},
@@ -89,14 +89,14 @@ func Test_filterNotAllowedPaths(t *testing.T) {
 		},
 		{
 			name: "Cleanup empty maps",
-			ctx: &filterIntentInput{
-				path: contract.Path{},
-				value: map[string]interface{}{
+			ctx: &FilterIntentInput{
+				Path: contract.Path{},
+				Value: map[string]interface{}{
 					"spec": map[string]interface{}{
 						"foo": "123",
 					},
 				},
-				shouldFilter: isNotAllowedPath(
+				ShouldFilter: IsNotAllowedPath(
 					[]contract.Path{}, // NOTE: we are filtering out everything not in this list (everything)
 				),
 			},
@@ -109,9 +109,9 @@ func Test_filterNotAllowedPaths(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			filterIntent(tt.ctx)
+			FilterIntent(tt.ctx)
 
-			g.Expect(tt.ctx.value).To(Equal(tt.wantValue))
+			g.Expect(tt.ctx.Value).To(Equal(tt.wantValue))
 		})
 	}
 }
@@ -119,14 +119,14 @@ func Test_filterNotAllowedPaths(t *testing.T) {
 func Test_filterIgnoredPaths(t *testing.T) {
 	tests := []struct {
 		name      string
-		ctx       *filterIntentInput
+		ctx       *FilterIntentInput
 		wantValue map[string]interface{}
 	}{
 		{
 			name: "Filters out ignore paths",
-			ctx: &filterIntentInput{
-				path: contract.Path{},
-				value: map[string]interface{}{
+			ctx: &FilterIntentInput{
+				Path: contract.Path{},
+				Value: map[string]interface{}{
 					"spec": map[string]interface{}{
 						"foo": "bar",
 						"controlPlaneEndpoint": map[string]interface{}{
@@ -135,7 +135,7 @@ func Test_filterIgnoredPaths(t *testing.T) {
 						},
 					},
 				},
-				shouldFilter: isIgnorePath(
+				ShouldFilter: IsIgnorePath(
 					[]contract.Path{
 						{"spec", "controlPlaneEndpoint"},
 					},
@@ -150,14 +150,14 @@ func Test_filterIgnoredPaths(t *testing.T) {
 		},
 		{
 			name: "Cleanup empty maps",
-			ctx: &filterIntentInput{
-				path: contract.Path{},
-				value: map[string]interface{}{
+			ctx: &FilterIntentInput{
+				Path: contract.Path{},
+				Value: map[string]interface{}{
 					"spec": map[string]interface{}{
 						"foo": "123",
 					},
 				},
-				shouldFilter: isIgnorePath(
+				ShouldFilter: IsIgnorePath(
 					[]contract.Path{
 						{"spec", "foo"},
 					},
@@ -169,16 +169,16 @@ func Test_filterIgnoredPaths(t *testing.T) {
 		},
 		{
 			name: "Cleanup empty nested maps",
-			ctx: &filterIntentInput{
-				path: contract.Path{},
-				value: map[string]interface{}{
+			ctx: &FilterIntentInput{
+				Path: contract.Path{},
+				Value: map[string]interface{}{
 					"spec": map[string]interface{}{
 						"bar": map[string]interface{}{
 							"foo": "123",
 						},
 					},
 				},
-				shouldFilter: isIgnorePath(
+				ShouldFilter: IsIgnorePath(
 					[]contract.Path{
 						{"spec", "bar", "foo"},
 					},
@@ -193,9 +193,9 @@ func Test_filterIgnoredPaths(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			filterIntent(tt.ctx)
+			FilterIntent(tt.ctx)
 
-			g.Expect(tt.ctx.value).To(Equal(tt.wantValue))
+			g.Expect(tt.ctx.Value).To(Equal(tt.wantValue))
 		})
 	}
 }
diff --git a/internal/util/ssa/managedfields.go b/internal/util/ssa/managedfields.go
index 06a715ce2f83..e56275468e5a 100644
--- a/internal/util/ssa/managedfields.go
+++ b/internal/util/ssa/managedfields.go
@@ -26,8 +26,75 @@ import (
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+
+	"sigs.k8s.io/cluster-api/internal/contract"
 )
 
+const classicManager = "manager"
+
+// DropManagedFields modifies the managedFields entries on the object that belong to "manager" (Operation=Update)
+// to drop ownership of the given paths if there is no field yet that is managed by `ssaManager`.
+//
+// If we want to be able to drop fields that were previously owned by the "manager" we have to ensure that
+// fields are not co-owned by "manager" and `ssaManager`. Otherwise, when we drop the fields with SSA
+// (i.e. `ssaManager`) the fields would remain as they are still owned by "manager".
+// The following code will do a one-time update on the managed fields.
+// We won't do this on subsequent reconciles. This case will be identified by checking if `ssaManager` owns any fields.
+// Dropping ownership in paths for existing "manager" entries (which could also be from other controllers) is safe,
+// as we assume that if other controllers are still writing fields on the object they will just do it again and thus
+// gain ownership again.
+func DropManagedFields(ctx context.Context, c client.Client, obj client.Object, ssaManager string, paths []contract.Path) error {
+	// Return if `ssaManager` already owns any fields.
+	if hasFieldsManagedBy(obj, ssaManager) {
+		return nil
+	}
+
+	// Since there is no field managed by `ssaManager` it means that
+	// this is the first time this object is being processed after the controller calling this function
+	// started to use SSA patches.
+	// It is required to clean-up managedFields from entries created by the regular patches.
+	// This will ensure that `ssaManager` will be able to modify the fields that
+	// were originally owned by "manager".
+	base := obj.DeepCopyObject().(client.Object)
+
+	// Modify managedFieldEntry for manager=manager and operation=update to drop ownership
+	// for the given paths to avoid having two managers holding values.
+	originalManagedFields := obj.GetManagedFields()
+	managedFields := make([]metav1.ManagedFieldsEntry, 0, len(originalManagedFields))
+	for _, managedField := range originalManagedFields {
+		if managedField.Manager == classicManager &&
+			managedField.Operation == metav1.ManagedFieldsOperationUpdate {
+			// Unmarshal the managed fields into a map[string]interface{}
+			fieldsV1 := map[string]interface{}{}
+			if err := json.Unmarshal(managedField.FieldsV1.Raw, &fieldsV1); err != nil {
+				return errors.Wrap(err, "failed to unmarshal managed fields")
+			}
+
+			// Filter out the ownership for the given paths.
+			FilterIntent(&FilterIntentInput{
+				Path:         contract.Path{},
+				Value:        fieldsV1,
+				ShouldFilter: IsIgnorePath(paths),
+			})
+
+			fieldsV1Raw, err := json.Marshal(fieldsV1)
+			if err != nil {
+				return errors.Wrap(err, "failed to marshal managed fields")
+			}
+			managedField.FieldsV1.Raw = fieldsV1Raw
+
+			managedFields = append(managedFields, managedField)
+		} else {
+			// Do not modify the entry. Use as is.
+			managedFields = append(managedFields, managedField)
+		}
+	}
+
+	obj.SetManagedFields(managedFields)
+
+	return c.Patch(ctx, obj, client.MergeFrom(base))
+}
+
 // CleanUpManagedFieldsForSSAAdoption deletes the managedFields entries on the object that belong to "manager" (Operation=Update)
 // if there is no field yet that is managed by `ssaManager`.
 // It adds an "empty" entry in managedFields of the object if no field is currently managed by `ssaManager`.
@@ -60,7 +127,7 @@ func CleanUpManagedFieldsForSSAAdoption(ctx context.Context, obj client.Object,
 	originalManagedFields := obj.GetManagedFields()
 	managedFields := make([]metav1.ManagedFieldsEntry, 0, len(originalManagedFields))
 	for i := range originalManagedFields {
-		if originalManagedFields[i].Manager == "manager" &&
+		if originalManagedFields[i].Manager == classicManager &&
 			originalManagedFields[i].Operation == metav1.ManagedFieldsOperationUpdate {
 			continue
 		}
diff --git a/internal/util/ssa/managedfields_test.go b/internal/util/ssa/managedfields_test.go
index c48aa536df3f..950895a44125 100644
--- a/internal/util/ssa/managedfields_test.go
+++ b/internal/util/ssa/managedfields_test.go
@@ -19,6 +19,7 @@ package ssa
 
 import (
 	"context"
+	"encoding/json"
 	"testing"
 
 	. "github.com/onsi/gomega"
@@ -26,13 +27,143 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"sigs.k8s.io/cluster-api/internal/contract"
 )
 
+func TestDropManagedFields(t *testing.T) {
+	ctx := context.Background()
+
+	ssaManager := "ssa-manager"
+
+	fieldV1Map := map[string]interface{}{
+		"f:metadata": map[string]interface{}{
+			"f:name":        map[string]interface{}{},
+			"f:labels":      map[string]interface{}{},
+			"f:annotations": map[string]interface{}{},
+			"f:finalizers":  map[string]interface{}{},
+		},
+	}
+	fieldV1, err := json.Marshal(fieldV1Map)
+	if err != nil {
+		panic(err)
+	}
+
+	objWithoutSSAManager := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "cm-1",
+			Namespace: "default",
+			ManagedFields: []metav1.ManagedFieldsEntry{{
+				Manager:    classicManager,
+				Operation:  metav1.ManagedFieldsOperationUpdate,
+				FieldsType: "FieldsV1",
+				FieldsV1:   &metav1.FieldsV1{Raw: fieldV1},
+			}},
+			Labels: map[string]string{
+				"label-1": "value-1",
+			},
+			Annotations: map[string]string{
+				"annotation-1": "value-1",
+			},
+			Finalizers: []string{"test-finalizer"},
+		},
+		Data: map[string]string{
+			"test-key": "test-value",
+		},
+	}
+
+	objectWithSSAManager := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "cm-2",
+			Namespace: "default",
+			ManagedFields: []metav1.ManagedFieldsEntry{
+				{
+					Manager:    classicManager,
+					Operation:  metav1.ManagedFieldsOperationUpdate,
+					FieldsType: "FieldsV1",
+					FieldsV1:   &metav1.FieldsV1{Raw: fieldV1},
+				},
+				{
+					Manager:   ssaManager,
+					Operation: metav1.ManagedFieldsOperationApply,
+				},
+			},
+			Labels: map[string]string{
+				"label-1": "value-1",
+			},
+			Annotations: map[string]string{
+				"annotation-1": "value-1",
+			},
+			Finalizers: []string{"test-finalizer"},
+		},
+		Data: map[string]string{
+			"test-key": "test-value",
+		},
+	}
+
+	tests := []struct {
+		name                string
+		obj                 client.Object
+		wantOwnershipToDrop bool
+	}{
+		{
+			name:                "should drop ownership of fields if there is no entry for ssaManager",
+			obj:                 objWithoutSSAManager,
+			wantOwnershipToDrop: true,
+		},
+		{
+			name:                "should not drop ownership of fields if there is an entry for ssaManager",
+			obj:                 objectWithSSAManager,
+			wantOwnershipToDrop: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+			fakeClient := fake.NewClientBuilder().WithObjects(tt.obj).Build()
+			labelsAndAnnotationsManagedFieldPaths := []contract.Path{
+				{"f:metadata", "f:annotations"},
+				{"f:metadata", "f:labels"},
+			}
+			g.Expect(DropManagedFields(ctx, fakeClient, tt.obj, ssaManager, labelsAndAnnotationsManagedFieldPaths)).To(Succeed())
+			if tt.wantOwnershipToDrop {
+				g.Expect(tt.obj.GetManagedFields()).ShouldNot(MatchFieldOwnership(
+					classicManager,
+					metav1.ManagedFieldsOperationUpdate,
+					contract.Path{"f:metadata", "f:labels"},
+				))
+				g.Expect(tt.obj.GetManagedFields()).ShouldNot(MatchFieldOwnership(
+					classicManager,
+					metav1.ManagedFieldsOperationUpdate,
+					contract.Path{"f:metadata", "f:annotations"},
+				))
+			} else {
+				g.Expect(tt.obj.GetManagedFields()).Should(MatchFieldOwnership(
+					classicManager,
+					metav1.ManagedFieldsOperationUpdate,
+					contract.Path{"f:metadata", "f:labels"},
+				))
+				g.Expect(tt.obj.GetManagedFields()).Should(MatchFieldOwnership(
+					classicManager,
+					metav1.ManagedFieldsOperationUpdate,
+					contract.Path{"f:metadata", "f:annotations"},
+				))
+			}
+			// Verify ownership of other fields is not affected.
+			g.Expect(tt.obj.GetManagedFields()).Should(MatchFieldOwnership(
+				classicManager,
+				metav1.ManagedFieldsOperationUpdate,
+				contract.Path{"f:metadata", "f:finalizers"},
+			))
+		})
+	}
+}
+
 func TestCleanUpManagedFieldsForSSAAdoption(t *testing.T) {
 	ctx := context.Background()
 
 	ssaManager := "ssa-manager"
-	classicManager := "manager"
 
 	objWithoutAnyManager := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
diff --git a/internal/util/ssa/matchers.go b/internal/util/ssa/matchers.go
index eb65c33e90bf..cdbffae91359 100644
--- a/internal/util/ssa/matchers.go
+++ b/internal/util/ssa/matchers.go
@@ -17,10 +17,14 @@ limitations under the License.
 package ssa
 
 import (
+	"encoding/json"
 	"fmt"
 
 	"github.com/onsi/gomega/types"
+	"github.com/pkg/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"sigs.k8s.io/cluster-api/internal/contract"
 )
 
 // MatchManagedFieldsEntry is a gomega Matcher to check if a ManagedFieldsEntry has the given name and operation.
@@ -56,3 +60,54 @@ func (mf *managedFieldMatcher) NegatedFailureMessage(actual interface{}) string
 	return fmt.Sprintf("Expected ManagedFieldsEntry to not match Manager:%s and Operation:%s, got Manager:%s, Operation:%s",
 		mf.manager, mf.operation, managedFieldsEntry.Manager, managedFieldsEntry.Operation)
 }
+
+// MatchFieldOwnership is a gomega Matcher to check if path is owned by the given manager and operation.
+// Note: The path has to be specified as is observed in managed fields. Example: to check if the labels are owned
+// by the correct manager the correct way to pass the path is contract.Path{"f:metadata","f:labels"}.
+func MatchFieldOwnership(manager string, operation metav1.ManagedFieldsOperationType, path contract.Path) types.GomegaMatcher {
+	return &fieldOwnershipMatcher{
+		path:      path,
+		manager:   manager,
+		operation: operation,
+	}
+}
+
+type fieldOwnershipMatcher struct {
+	path      contract.Path
+	manager   string
+	operation metav1.ManagedFieldsOperationType
+}
+
+func (fom *fieldOwnershipMatcher) Match(actual interface{}) (bool, error) {
+	managedFields, ok := actual.([]metav1.ManagedFieldsEntry)
+	if !ok {
+		return false, fmt.Errorf("expecting []metav1.ManagedFieldsEntry got %T", actual)
+	}
+	for _, managedFieldsEntry := range managedFields {
+		if managedFieldsEntry.Manager == fom.manager && managedFieldsEntry.Operation == fom.operation {
+			fieldsV1 := map[string]interface{}{}
+			if err := json.Unmarshal(managedFieldsEntry.FieldsV1.Raw, &fieldsV1); err != nil {
+				return false, errors.Wrap(err, "failed to parse managedFieldsEntry.FieldsV1")
+			}
+			FilterIntent(&FilterIntentInput{
+				Path:         contract.Path{},
+				Value:        fieldsV1,
+				ShouldFilter: IsNotAllowedPath([]contract.Path{fom.path}),
+			})
+			return len(fieldsV1) > 0, nil
+		}
+	}
+	return false, nil
+}
+
+func (fom *fieldOwnershipMatcher) FailureMessage(actual interface{}) string {
+	managedFields := actual.([]metav1.ManagedFieldsEntry)
+	return fmt.Sprintf("Expected Path %s to be owned by Manager:%s and Operation:%s, did not find correct ownership: %s",
+		fom.path, fom.manager, fom.operation, managedFields)
+}
+
+func (fom *fieldOwnershipMatcher) NegatedFailureMessage(actual interface{}) string {
+	managedFields := actual.([]metav1.ManagedFieldsEntry)
+	return fmt.Sprintf("Expected Path %s to not be owned by Manager:%s and Operation:%s, did not find correct ownership: %s",
+		fom.path, fom.manager, fom.operation, managedFields)
+}