Radius Controller May Delete a Container Resource via an Injected Deployment Annotation (Multi-Tenant Installs)

@@ -53,7 +53,6 @@ data:
     tracerProvider:
       enabled: true
       serviceName: "controller"
-      zipkin: 
+      zipkin:
         url: {{ .Values.global.zipkin.url }}
     {{- end }}
-    
\ No newline at end of file

@@ -59,8 +59,8 @@ spec:
       containers:
       - name: controller
         image: "{{ include "radius.image" (dict "image" .Values.controller.image "tag" (.Values.controller.tag | default .Values.global.imageTag | default $appversion) "global" .Values.global) }}"
-        imagePullPolicy: 'Always'
-        args: 
+        imagePullPolicy: {{ .Values.controller.imagePullPolicy | default "Always" | quote }}
+        args:
         - '--config-file'
         - '/etc/config/controller-config.yaml'
         env:
@@ -132,4 +132,4 @@ spec:
         {{- if .Values.global.appendRootCA.cert }}
         - name: ssl-certs
           emptyDir: {}
-        {{- end }}
\ No newline at end of file
+        {{- end }}

@@ -93,4 +93,4 @@ roleRef:
 subjects:
 - kind: ServiceAccount
   name: controller
-  namespace: {{ .Release.Namespace }}
\ No newline at end of file
+  namespace: {{ .Release.Namespace }}

@@ -13,4 +13,4 @@ spec:
       protocol: TCP
       targetPort: 9443
   selector:
-      app.kubernetes.io/name: controller
+    app.kubernetes.io/name: controller

@@ -44,4 +44,4 @@ webhooks:
     - UPDATE
     resources:
     - recipes
-  sideEffects: None
\ No newline at end of file
+  sideEffects: None

@@ -0,0 +1,41 @@
+{{- if .Values.rp.security.annotationProtection.enabled }}
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingAdmissionPolicy
+metadata:
+  name: radius-rp-annotation-protection
+  labels:
+    app.kubernetes.io/name: applications-rp
+    app.kubernetes.io/part-of: radius
+spec:
+  failurePolicy: Fail
+  matchConstraints:
+    resourceRules:
+    - apiGroups: ["apps"]
+      apiVersions: ["v1"]
+      operations: ["UPDATE"]
+      resources: ["deployments"]
+  validations:
+  - expression: >-
+      request.userInfo.username == "system:serviceaccount:{{ .Release.Namespace }}:{{ .Values.rp.security.annotationProtection.allowedServiceAccountName }}" ||
+      (
+        (!has(oldObject.metadata.annotations) || !oldObject.metadata.annotations.exists(k, k == "radapp.io/status")) &&
+        (!has(object.metadata.annotations) || !object.metadata.annotations.exists(k, k == "radapp.io/status"))
+      ) ||
+      (
+        has(oldObject.metadata.annotations) && oldObject.metadata.annotations.exists(k, k == "radapp.io/status") &&
+        has(object.metadata.annotations) && object.metadata.annotations.exists(k, k == "radapp.io/status") &&
+        object.metadata.annotations["radapp.io/status"] == oldObject.metadata.annotations["radapp.io/status"]
+      )
+    message: "modifying annotation radapp.io/status is restricted to the Radius controller service account system:serviceaccount:{{ .Release.Namespace }}:{{ .Values.rp.security.annotationProtection.allowedServiceAccountName }}"
+---
+apiVersion: admissionregistration.k8s.io/v1
+kind: ValidatingAdmissionPolicyBinding
+metadata:
+  name: radius-rp-annotation-protection-binding
+  labels:
+    app.kubernetes.io/name: applications-rp
+    app.kubernetes.io/part-of: radius
+spec:
+  policyName: radius-rp-annotation-protection
+  validationActions: [Deny]
+{{- end }}

@@ -50,7 +50,7 @@ data:
     tracerProvider:
       enabled: true
       serviceName: "applications-rp"
-      zipkin: 
+      zipkin:
         url: {{ .Values.global.zipkin.url }}
     {{- end }}
     bicep:

@@ -64,11 +64,11 @@ spec:
           echo "Terraform init container starting..."
           echo "Running on architecture: $(uname -m)"
           echo "Alpine version: $(cat /etc/alpine-release)"
-          
+
           # Create terraform directory
           mkdir -p "{{ .Values.rp.terraform.path }}"
           echo "Created directory: {{ .Values.rp.terraform.path }}"
-          
+
           # Detect architecture for terraform download
           ARCH=$(uname -m)
           case $ARCH in
@@ -77,16 +77,16 @@ spec:
             *) echo "ERROR: Unsupported architecture: $ARCH"; exit 1 ;;
           esac
           echo "Terraform architecture: $TERRAFORM_ARCH"
-          
+
           # Install wget and unzip if not available (Alpine doesn't include them by default)
           if ! which wget >/dev/null 2>&1 || ! which unzip >/dev/null 2>&1; then
             echo "Installing wget and unzip..."
             apk add --no-cache wget unzip || { echo "ERROR: Failed to install wget/unzip"; exit 3; }
           fi
-          
+
           # Determine download URL
           TERRAFORM_URL="{{ .Values.global.terraform.downloadUrl }}"
-          
+
           # If no custom URL provided, fetch latest version from HashiCorp API
           if [[ -z "$TERRAFORM_URL" ]]; then
             echo "Fetching latest Terraform version from HashiCorp API..."
@@ -98,36 +98,36 @@ spec:
             echo "Latest Terraform version: $LATEST_VERSION"
             TERRAFORM_URL="https://releases.hashicorp.com/terraform/${LATEST_VERSION}/terraform_${LATEST_VERSION}_linux_${TERRAFORM_ARCH}.zip"
           fi
-          
+
           echo "Download URL: $TERRAFORM_URL"
-          
+
           # Basic connectivity and environment info
           echo "Environment debug info:"
           echo "- Date: $(date)"
           echo "- PWD: $(pwd)"
           echo "- Available commands: $(which wget || echo 'no wget') $(which unzip || echo 'no unzip')"
-          
+
           # Install wget and unzip if not available (Alpine doesn't include them by default)
           if ! which wget >/dev/null 2>&1 || ! which unzip >/dev/null 2>&1; then
             echo "Installing wget and unzip..."
             apk add --no-cache wget unzip || { echo "ERROR: Failed to install wget/unzip"; exit 3; }
           fi
-          
+
           # Download and extract terraform using wget
           cd /tmp
           echo "Downloading terraform using wget..."
           wget "${TERRAFORM_URL}" -O terraform.zip || { echo "ERROR: Failed to download terraform"; exit 4; }
-          
+
           echo "Extracting terraform using unzip..."
           unzip terraform.zip || { echo "ERROR: Failed to extract terraform"; exit 5; }
-          
+
           echo "Installing terraform binary..."
           cp terraform "{{ .Values.rp.terraform.path }}/terraform" || { echo "ERROR: Failed to copy terraform"; exit 6; }
           chmod +x "{{ .Values.rp.terraform.path }}/terraform" || { echo "ERROR: Failed to make terraform executable"; exit 7; }
-          
+
           # Create marker file to indicate pre-mounted binary is available
           echo "pre-mounted" > "{{ .Values.rp.terraform.path }}/.terraform-source"
-          
+
           echo "Terraform binary successfully pre-downloaded and installed"
         volumeMounts:
         - name: terraform

@@ -141,6 +141,14 @@ rp:
     deleteRetryDelaySeconds: 60
   terraform:
     path: "/terraform"
+  security:
+    annotationProtection:
+      # Enable ValidatingAdmissionPolicy to protect radapp.io/status from user tampering.
+      # Keep disabled by default for compatibility with clusters that do not support
+      # admissionregistration.k8s.io/v1 ValidatingAdmissionPolicy.
+      enabled: false
+      # Service account allowed to modify protected annotations.
+      allowedServiceAccountName: "applications-rp"
 
 dashboard:
   enabled: true

@@ -85,9 +85,9 @@ func ExtractResourceTypes(template map[string]any) []ResourceTypeEntry {
 		}
 
 		entry := ResourceTypeEntry{FullType: resourceType}
-		if idx := strings.Index(resourceType, "@"); idx >= 0 {
-			entry.Type = resourceType[:idx]
-			entry.APIVersion = resourceType[idx+1:]
+		if before, after, ok0 := strings.Cut(resourceType, "@"); ok0 {
+			entry.Type = before
+			entry.APIVersion = after
 		} else {
 			entry.Type = resourceType
 		}

@@ -200,6 +200,7 @@ func Test_Run(t *testing.T) {
 
 		err = runner.Run(context.Background())
 		require.NoError(t, err)
+
 		require.Equal(t, expectedOutput, outputSink.Writes)
 	})
 

@@ -228,7 +228,7 @@ func normalizeRecipePacks(recipepacks []string) []string {
 	seen := map[string]struct{}{}
 	result := []string{}
 	for _, value := range recipepacks {
-		for _, p := range strings.Split(value, ",") {
+		for p := range strings.SplitSeq(value, ",") {
 			trimmed := strings.TrimSpace(p)
 			if trimmed == "" {
 				continue

@@ -1150,7 +1150,7 @@ type cloudProviderPromptMatcher struct {
 }
 
 // Matches implements gomock.Matcher
-func (*cloudProviderPromptMatcher) Matches(x interface{}) bool {
+func (*cloudProviderPromptMatcher) Matches(x any) bool {
 	return x == confirmCloudProviderPrompt || x == confirmCloudProviderAdditionalPrompt
 }
 

@@ -512,7 +512,7 @@ func Test_Run(t *testing.T) {
 			appManagementClient.EXPECT().
 				GetResource(gomock.Any(), "Applications.Core/containers", "test-container").
 				Return(generated.GenericResource{
-					Properties: map[string]interface{}{
+					Properties: map[string]any{
 						"environment": "/planes/radius/local/resourceGroups/test-group/providers/Applications.Core/environments/my-test-env",
 						"application": "/planes/radius/local/resourceGroups/test-group/providers/Applications.Core/applications/my-test-app",
 					},

@@ -24,6 +24,7 @@ import (
 	"strings"
 
 	radappiov1alpha3 "github.com/radius-project/radius/pkg/controller/api/radapp.io/v1alpha3"
+	"github.com/radius-project/radius/pkg/ucp/resources"
 	appsv1 "k8s.io/api/apps/v1"
 )
 
@@ -115,6 +116,11 @@ func readAnnotations(deployment *appsv1.Deployment) (deploymentAnnotations, erro
 			return result, fmt.Errorf("failed to unmarshal status annotation: %w", err)
 		}
 
+		err = validateDeploymentStatus(&s)
+		if err != nil {
+			return result, fmt.Errorf("invalid status annotation: %w", err)
+		}
+
 		result.Status = &s
 	}
 
@@ -140,6 +146,41 @@ func readAnnotations(deployment *appsv1.Deployment) (deploymentAnnotations, erro
 	return result, nil
 }
 
+func validateDeploymentStatus(status *deploymentStatus) error {
+	if status == nil {
+		return nil
+	}
+
+	// The reconciler state machine may persist status.scope (and an in-progress operation) before
+	// status.container is known, so scope-only status is valid. When the environment or application
+	// changes, status.scope is updated to the new scope while status.container still references the
+	// previous container (which the reconciler then deletes), so we do not require status.scope to
+	// match the container's root scope.
+	if status.Scope != "" {
+		if _, err := resources.ParseScope(status.Scope); err != nil {
+			return fmt.Errorf("invalid status.scope: %w", err)
+		}
+	}
+
+	if status.Container == "" {
+		return nil
+	}
+
+	if status.Scope == "" {
+		return fmt.Errorf("status.scope must be set when status.container is set")
+	}
+
+	parsedContainer, err := resources.ParseResource(status.Container)
+	if err != nil {
+		return fmt.Errorf("invalid status.container: %w", err)
+	}
+	if !strings.EqualFold(parsedContainer.Type(), applicationsCoreContainersResourceType) {
+		return fmt.Errorf("status.container type %q is not %q", parsedContainer.Type(), applicationsCoreContainersResourceType)
+	}
+
+	return nil
+}
+
 // ApplyToDeployment applies the configuration and status to a Deployment.
 //
 // This should be used before saving the Deployment's state.

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/radius-project/radius/pkg/ucp/resources"
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -31,7 +32,7 @@ func Test_readAnnotations(t *testing.T) {
 		Scope:       "/planes/radius/local/resourceGroups/controller-test",
 		Application: "test-application",
 		Environment: "test-environment",
-		Container:   "test-container",
+		Container:   "/planes/radius/local/resourceGroups/controller-test/providers/Applications.Core/containers/test-container",
 		Operation:   nil,
 		Phrase:      deploymentPhraseReady,
 	}
@@ -43,6 +44,9 @@ func Test_readAnnotations(t *testing.T) {
 	// so that an unmarshaling error can be triggered.
 	invalidDeploymentStatus := []byte(`{"invalid": "json"`)
 
+	_, invalidContainerIDErr := resources.ParseResource("not-a-resource-id")
+	_, invalidScopeIDErr := resources.ParseScope("not-a-scope")
+
 	tests := []struct {
 		name        string
 		deployment  *appsv1.Deployment
@@ -156,11 +160,101 @@ func Test_readAnnotations(t *testing.T) {
 			err: fmt.Errorf("failed to unmarshal status annotation: %w",
 				json.Unmarshal(invalidDeploymentStatus, &deploymentStatus{})),
 		},
+		{
+			name: "status-invalid-container-id",
+			deployment: &appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AnnotationRadiusStatus: `{"scope":"/planes/radius/local/resourceGroups/controller-test","container":"not-a-resource-id"}`,
+					},
+				},
+			},
+			annotations: deploymentAnnotations{ConfigurationHash: ""},
+			err:         fmt.Errorf("invalid status annotation: invalid status.container: %w", invalidContainerIDErr),
+		},
+		{
+			name: "status-scope-container-mismatch-allowed",
+			deployment: &appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AnnotationRadiusStatus: `{"scope":"/planes/radius/local/resourceGroups/controller-test","container":"/planes/radius/local/resourceGroups/other/providers/Applications.Core/containers/test-container"}`,
+					},
+				},
+			},
+			// The reconciler intentionally produces this transitional state when the environment or
+			// application changes: status.scope advances to the new scope while status.container still
+			// references the previous container until it is deleted.
+			annotations: deploymentAnnotations{
+				ConfigurationHash: "",
+				Status: &deploymentStatus{
+					Scope:     "/planes/radius/local/resourceGroups/controller-test",
+					Container: "/planes/radius/local/resourceGroups/other/providers/Applications.Core/containers/test-container",
+				},
+			},
+			err: nil,
+		},
+		{
+			name: "status-container-wrong-resource-type",
+			deployment: &appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AnnotationRadiusStatus: `{"scope":"/planes/radius/local/resourceGroups/controller-test","container":"/planes/radius/local/resourceGroups/controller-test/providers/Applications.Core/applications/test-app"}`,
+					},
+				},
+			},
+			annotations: deploymentAnnotations{ConfigurationHash: ""},
+			err:         fmt.Errorf("invalid status annotation: status.container type %q is not %q", "Applications.Core/applications", applicationsCoreContainersResourceType),
+		},
+		{
+			name: "status-only-scope-set",
+			deployment: &appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AnnotationRadiusStatus: `{"scope":"/planes/radius/local/resourceGroups/controller-test"}`,
+					},
+				},
+			},
+			annotations: deploymentAnnotations{
+				ConfigurationHash: "",
+				Status: &deploymentStatus{
+					Scope: "/planes/radius/local/resourceGroups/controller-test",
+				},
+			},
+			err: nil,
+		},
+		{
+			name: "status-only-container-set",
+			deployment: &appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AnnotationRadiusStatus: `{"container":"/planes/radius/local/resourceGroups/controller-test/providers/Applications.Core/containers/test-container"}`,
+					},
+				},
+			},
+			annotations: deploymentAnnotations{ConfigurationHash: ""},
+			err:         fmt.Errorf("invalid status annotation: status.scope must be set when status.container is set"),
+		},
+		{
+			name: "status-invalid-scope-only",
+			deployment: &appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						AnnotationRadiusStatus: `{"scope":"not-a-scope"}`,
+					},
+				},
+			},
+			annotations: deploymentAnnotations{ConfigurationHash: ""},
+			err:         fmt.Errorf("invalid status annotation: invalid status.scope: %w", invalidScopeIDErr),
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			annotations, err := readAnnotations(tt.deployment)
-			require.Equal(t, tt.err, err)
+			if tt.err == nil {
+				require.NoError(t, err)
+			} else {
+				require.EqualError(t, err, tt.err.Error())
+			}
 			require.Equal(t, tt.annotations, annotations)
 		})
 	}

@@ -45,6 +45,16 @@ const (
 	// DeploymentFinalizer is the name of the finalizer added to Deployments.
 	DeploymentFinalizer = "radapp.io/deployment-finalizer"
 
+	// EventScopeMismatch is emitted when a deployment annotation references a scope outside
+	// of the deployment's ownership boundary.
+	EventScopeMismatch = "ScopeMismatch"
+
+	// EventContainerOwnershipMismatch is emitted when a container does not reference
+	// the source Kubernetes deployment resource.
+	EventContainerOwnershipMismatch = "ContainerOwnershipMismatch"
+
+	applicationsCoreContainersResourceType = "Applications.Core/containers"
+
 	// RecipeFinalizer is the name of the finalizer added to Recipes.
 	RecipeFinalizer = "radapp.io/recipe-finalizer"
 

@@ -363,7 +363,7 @@ func (r *DeploymentReconciler) reconcileUpdate(ctx context.Context, deployment *
 func (r *DeploymentReconciler) reconcileDelete(ctx context.Context, deployment *appsv1.Deployment, annotations *deploymentAnnotations) (ctrl.Result, error) {
 	logger := ucplog.FromContextOrDiscard(ctx)
 
-	poller, err := r.startDeleteOperationIfNeeded(ctx, annotations)
+	poller, err := r.startDeleteOperationIfNeeded(ctx, deployment, annotations)
 	if err != nil {
 		logger.Error(err, "Unable to delete resource.")
 		r.EventRecorder.Event(deployment, corev1.EventTypeWarning, "ResourceError", err.Error())
@@ -415,8 +415,7 @@ func (r *DeploymentReconciler) startPutOrDeleteOperationIfNeeded(ctx context.Con
 		// the old resource and create a new one.
 		logger.Info("Container is already created but is out-of-date")
 
-		logger.Info("Starting DELETE operation.")
-		poller, err := deleteContainer(ctx, r.Radius, annotations.Status.Container)
+		poller, err := r.startDeleteOperationIfNeeded(ctx, deployment, annotations)
 		if err != nil {
 			return nil, nil, false, err
 		} else if poller != nil {
@@ -480,13 +479,42 @@ func (r *DeploymentReconciler) startPutOrDeleteOperationIfNeeded(ctx context.Con
 	return poller, nil, false, nil
 }
 
-func (r *DeploymentReconciler) startDeleteOperationIfNeeded(ctx context.Context, annotations *deploymentAnnotations) (sdkclients.Poller[v20231001preview.ContainersClientDeleteResponse], error) {
+func (r *DeploymentReconciler) startDeleteOperationIfNeeded(ctx context.Context, deployment *appsv1.Deployment, annotations *deploymentAnnotations) (sdkclients.Poller[v20231001preview.ContainersClientDeleteResponse], error) {
 	logger := ucplog.FromContextOrDiscard(ctx)
+
+	if annotations == nil || annotations.Status == nil {
+		logger.Info("Container is already deleted (or was never created).")
+		return nil, nil
+	}
+
 	if annotations.Status.Container == "" {
 		logger.Info("Container is already deleted (or was never created).")
 		return nil, nil
 	}
 
+	expectedDeploymentResourceID := makeKubernetesDeploymentResourceID(deployment.Namespace, deployment.Name)
+	container, err := fetchContainerResource(ctx, r.Radius, annotations.Status.Container)
+	if clients.Is404Error(err) {
+		logger.Info("Container was already deleted before cleanup began.", "container", annotations.Status.Container)
+		annotations.Status.Container = ""
+		return nil, nil
+	} else if err != nil {
+		return nil, fmt.Errorf("failed to fetch container before delete: %w", err)
+	}
+
+	if !containerHasResourceReference(&container, expectedDeploymentResourceID) {
+		logger.Info("Refusing cross-ownership delete attempt.",
+			"container", annotations.Status.Container,
+			"expectedDeploymentResource", expectedDeploymentResourceID,
+		)
+		r.EventRecorder.Event(deployment, corev1.EventTypeWarning, EventContainerOwnershipMismatch,
+			fmt.Sprintf("Container %q does not reference deployment resource %q", annotations.Status.Container, expectedDeploymentResourceID))
+
+		// Clear status to avoid a persistent reconcile loop on a tampered annotation.
+		annotations.Status.Container = ""
+		return nil, nil
+	}
+
 	logger.Info("Starting DELETE operation.")
 	poller, err := deleteContainer(ctx, r.Radius, annotations.Status.Container)
 	if err != nil {

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"testing"
 	"time"
 
@@ -34,7 +35,9 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	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/types"
+	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	crconfig "sigs.k8s.io/controller-runtime/pkg/config"
@@ -95,6 +98,118 @@ func SetupDeploymentTest(t *testing.T) (*mockRadiusClient, client.Client) {
 	return radius, mgr.GetClient()
 }
 
+func Test_DeploymentReconciler_StartDeleteOperationIfNeeded_OwnershipMismatch_BlocksDelete(t *testing.T) {
+	ctx := testcontext.New(t)
+	radius := NewMockRadiusClient()
+	reconciler := &DeploymentReconciler{
+		Radius:        radius,
+		EventRecorder: record.NewFakeRecorder(10),
+	}
+
+	deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: "current-app", Namespace: "current-namespace"}}
+	containerID := "/planes/radius/local/resourceGroups/tenant-b/providers/Applications.Core/containers/other-container"
+
+	radius.Update(func() {
+		radius.containers[containerID] = v20231001preview.ContainerResource{
+			Properties: &v20231001preview.ContainerProperties{
+				Resources: []*v20231001preview.ResourceReference{{
+					ID: new("/planes/kubernetes/local/namespaces/other-namespace/providers/apps/Deployment/other-app"),
+				}},
+			},
+		}
+	})
+
+	annotations := &deploymentAnnotations{Status: &deploymentStatus{Container: containerID}}
+	poller, err := reconciler.startDeleteOperationIfNeeded(ctx, deployment, annotations)
+	require.NoError(t, err)
+	require.Nil(t, poller)
+	require.Empty(t, annotations.Status.Container)
+	requireNoDeleteOperation(t, radius, containerID)
+
+	_, err = radius.Containers("/planes/radius/local/resourceGroups/tenant-b").Get(ctx, "other-container", nil)
+	require.NoError(t, err)
+}
+
+func Test_DeploymentReconciler_StartPutOrDeleteOperationIfNeeded_OwnershipMismatch_BlocksDelete(t *testing.T) {
+	ctx := testcontext.New(t)
+	radius := NewMockRadiusClient()
+	reconciler := &DeploymentReconciler{
+		Radius:        radius,
+		EventRecorder: record.NewFakeRecorder(10),
+	}
+
+	deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: "current-app", Namespace: "current-namespace"}}
+	otherContainerID := "/planes/radius/local/resourceGroups/tenant-b/providers/Applications.Core/containers/other-container"
+
+	radius.Update(func() {
+		radius.containers[otherContainerID] = v20231001preview.ContainerResource{
+			Properties: &v20231001preview.ContainerProperties{
+				Resources: []*v20231001preview.ResourceReference{{
+					ID: new("/planes/kubernetes/local/namespaces/other-namespace/providers/apps/Deployment/other-container"),
+				}},
+			},
+		}
+	})
+
+	annotations := &deploymentAnnotations{
+		Configuration: &deploymentConfiguration{},
+		Status: &deploymentStatus{
+			Scope:       "/planes/radius/local/resourceGroups/tenant-a",
+			Application: "/planes/radius/local/resourceGroups/tenant-a/providers/Applications.Core/applications/current-app",
+			Container:   otherContainerID,
+		},
+	}
+
+	updatePoller, deletePoller, waiting, err := reconciler.startPutOrDeleteOperationIfNeeded(ctx, deployment, annotations)
+	require.NoError(t, err)
+	require.NotNil(t, updatePoller)
+	require.Nil(t, deletePoller)
+	require.False(t, waiting)
+	requireNoDeleteOperation(t, radius, otherContainerID)
+
+	_, err = radius.Containers("/planes/radius/local/resourceGroups/tenant-b").Get(ctx, "other-container", nil)
+	require.NoError(t, err)
+}
+
+func Test_DeploymentReconciler_StartDeleteOperationIfNeeded_OwnershipMatch_AllowsDelete(t *testing.T) {
+	ctx := testcontext.New(t)
+	radius := NewMockRadiusClient()
+	reconciler := &DeploymentReconciler{
+		Radius:        radius,
+		EventRecorder: record.NewFakeRecorder(10),
+	}
+
+	deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Name: "current-app", Namespace: "current-namespace"}}
+	containerID := "/planes/radius/local/resourceGroups/tenant-a/providers/Applications.Core/containers/current-app"
+
+	radius.Update(func() {
+		radius.containers[containerID] = v20231001preview.ContainerResource{
+			Properties: &v20231001preview.ContainerProperties{
+				Resources: []*v20231001preview.ResourceReference{{
+					ID: new(makeKubernetesDeploymentResourceID("current-namespace", "current-app")),
+				}},
+			},
+		}
+	})
+
+	annotations := &deploymentAnnotations{Status: &deploymentStatus{Container: containerID}}
+	poller, err := reconciler.startDeleteOperationIfNeeded(ctx, deployment, annotations)
+	require.NoError(t, err)
+	require.NotNil(t, poller)
+	require.Equal(t, containerID, annotations.Status.Container)
+}
+
+func requireNoDeleteOperation(t *testing.T, radius *mockRadiusClient, resourceID string) {
+	t.Helper()
+
+	radius.lock.Lock()
+	defer radius.lock.Unlock()
+
+	for _, operation := range radius.operations {
+		require.False(t, operation.Kind == http.MethodDelete && operation.ResourceID == resourceID, "unexpected delete operation for %s", resourceID)
+	}
+}
+
 // Creates a deployment with Radius enabled.
 //
 // Then exercises the cleanup path by deleting the deployment.

@@ -226,7 +226,7 @@ func fetchResource(ctx context.Context, radius RadiusClient, resourceID string)
 }
 
 func deleteContainer(ctx context.Context, radius RadiusClient, containerID string) (sdkclients.Poller[corerpv20231001preview.ContainersClientDeleteResponse], error) {
-	id, err := resources.Parse(containerID)
+	id, err := parseContainerResourceID(containerID)
 	if err != nil {
 		return nil, err
 	}
@@ -252,8 +252,44 @@ func deleteContainer(ctx context.Context, radius RadiusClient, containerID strin
 	return nil, nil
 }
 
+func makeKubernetesDeploymentResourceID(namespace string, name string) string {
+	return "/planes/kubernetes/local/namespaces/" + namespace + "/providers/apps/Deployment/" + name
+}
+
+func containerHasResourceReference(container *corerpv20231001preview.ContainerResource, expectedResourceID string) bool {
+	if container == nil || container.Properties == nil {
+		return false
+	}
+
+	for _, resource := range container.Properties.Resources {
+		if resource == nil || resource.ID == nil {
+			continue
+		}
+
+		if strings.EqualFold(*resource.ID, expectedResourceID) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func fetchContainerResource(ctx context.Context, radius RadiusClient, containerID string) (corerpv20231001preview.ContainerResource, error) {
+	id, err := parseContainerResourceID(containerID)
+	if err != nil {
+		return corerpv20231001preview.ContainerResource{}, err
+	}
+
+	response, err := radius.Containers(id.RootScope()).Get(ctx, id.Name(), nil)
+	if err != nil {
+		return corerpv20231001preview.ContainerResource{}, err
+	}
+
+	return response.ContainerResource, nil
+}
+
 func createOrUpdateContainer(ctx context.Context, radius RadiusClient, containerID string, properties *corerpv20231001preview.ContainerProperties) (sdkclients.Poller[corerpv20231001preview.ContainersClientCreateOrUpdateResponse], error) {
-	id, err := resources.Parse(containerID)
+	id, err := parseContainerResourceID(containerID)
 	if err != nil {
 		return nil, err
 	}
@@ -284,6 +320,19 @@ func createOrUpdateContainer(ctx context.Context, radius RadiusClient, container
 	return nil, nil
 }
 
+func parseContainerResourceID(containerID string) (resources.ID, error) {
+	id, err := resources.ParseResource(containerID)
+	if err != nil {
+		return resources.ID{}, err
+	}
+
+	if !strings.EqualFold(id.Type(), applicationsCoreContainersResourceType) {
+		return resources.ID{}, fmt.Errorf("resource type %q is not %q", id.Type(), applicationsCoreContainersResourceType)
+	}
+
+	return id, nil
+}
+
 func generateDeploymentResourceName(resourceId string) (string, error) {
 	id, err := resources.ParseResource(resourceId)
 	if err != nil {

@@ -3,6 +3,8 @@ package reconciler
 import (
 	"testing"
 
+	v20231001preview "github.com/radius-project/radius/pkg/corerp/api/v20231001preview"
+	"github.com/radius-project/radius/pkg/to"
 	"github.com/stretchr/testify/require"
 )
 
@@ -86,3 +88,55 @@ func TestConvertToARMJSONParameters(t *testing.T) {
 		})
 	}
 }
+
+func TestMakeKubernetesDeploymentResourceID(t *testing.T) {
+	got := makeKubernetesDeploymentResourceID("current-namespace", "current-app")
+	require.Equal(t, "/planes/kubernetes/local/namespaces/current-namespace/providers/apps/Deployment/current-app", got)
+}
+
+func TestContainerHasResourceReference(t *testing.T) {
+	tests := []struct {
+		name             string
+		container        *v20231001preview.ContainerResource
+		expectedResource string
+		want             bool
+	}{
+		{
+			name: "has matching resource reference",
+			container: &v20231001preview.ContainerResource{
+				Properties: &v20231001preview.ContainerProperties{
+					Resources: []*v20231001preview.ResourceReference{{
+						ID: to.Ptr("/planes/kubernetes/local/namespaces/current-namespace/providers/apps/Deployment/current-app"),
+					}},
+				},
+			},
+			expectedResource: "/planes/kubernetes/local/namespaces/current-namespace/providers/apps/Deployment/current-app",
+			want:             true,
+		},
+		{
+			name: "missing matching reference",
+			container: &v20231001preview.ContainerResource{
+				Properties: &v20231001preview.ContainerProperties{
+					Resources: []*v20231001preview.ResourceReference{{
+						ID: to.Ptr("/planes/kubernetes/local/namespaces/other-namespace/providers/apps/Deployment/other-app"),
+					}},
+				},
+			},
+			expectedResource: "/planes/kubernetes/local/namespaces/current-namespace/providers/apps/Deployment/current-app",
+			want:             false,
+		},
+		{
+			name:             "nil container",
+			container:        nil,
+			expectedResource: "/planes/kubernetes/local/namespaces/current-namespace/providers/apps/Deployment/current-app",
+			want:             false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := containerHasResourceReference(tt.container, tt.expectedResource)
+			require.Equal(t, tt.want, got)
+		})
+	}
+}

@@ -213,9 +213,7 @@ func getConfigurationV20250801(ctx context.Context, environment *v20250801previe
 		// registry credentials (key=`token`).
 		if len(tfProps.Terraformrc.Credentials) > 0 {
 			config.RecipeConfig.Terraform.Credentials = make(map[string]datamodel.TerraformCredentialConfig, len(tfProps.Terraformrc.Credentials))
-			for host, cred := range tfProps.Terraformrc.Credentials {
-				config.RecipeConfig.Terraform.Credentials[host] = cred
-			}
+			maps.Copy(config.RecipeConfig.Terraform.Credentials, tfProps.Terraformrc.Credentials)
 		}
 
 		// Map env vars into the legacy shape.