diff --git a/console-web/app/(dashboard)/tenants/new/page.tsx b/console-web/app/(dashboard)/tenants/new/page.tsx
index 927e9e8..325bee8 100644
--- a/console-web/app/(dashboard)/tenants/new/page.tsx
+++ b/console-web/app/(dashboard)/tenants/new/page.tsx
@@ -243,11 +243,21 @@ export default function TenantCreatePage() {
           const user = asRecord(item)
           const userName = asString(user?.name)
           const userPolicies = asStringArray(user?.policies)
-          if (!user || !userName || !userPolicies || userPolicies.length === 0) {
+          const rawCredsSecret = user?.credsSecret ?? user?.creds_secret
+          const credsSecret = asRecord(rawCredsSecret)
+          const credsSecretName = asString(credsSecret?.name)
+          if (
+            !user ||
+            !userName ||
+            !userPolicies ||
+            userPolicies.length === 0 ||
+            (rawCredsSecret != null && !credsSecretName)
+          ) {
             throw new Error(t("YAML user provisioning fields are invalid"))
           }
           return {
             name: userName,
+            credsSecret: credsSecretName ? { name: credsSecretName } : undefined,
             policies: userPolicies,
           }
         })
diff --git a/console-web/types/api.ts b/console-web/types/api.ts
index cff5cbe..bb73ef5 100644
--- a/console-web/types/api.ts
+++ b/console-web/types/api.ts
@@ -115,6 +115,9 @@ export interface ProvisioningPolicy {
 
 export interface ProvisioningUser {
   name: string
+  credsSecret?: {
+    name: string
+  }
   policies: string[]
   deletionPolicy?: ProvisioningDeletionPolicy
 }
@@ -136,6 +139,8 @@ export interface ProvisioningItemStatus {
   lastAppliedHash?: string | null
   lastAppliedGeneration?: number | null
   observedSecretResourceVersion?: string | null
+  observedSecretName?: string | null
+  lastAppliedAccessKeyHash?: string | null
   policies?: string[]
   region?: string | null
   objectLock?: boolean | null
diff --git a/deploy/rustfs-operator/README.md b/deploy/rustfs-operator/README.md
index 358623d..978fa7c 100755
--- a/deploy/rustfs-operator/README.md
+++ b/deploy/rustfs-operator/README.md
@@ -150,6 +150,8 @@ spec:
           key: policy.json
   users:
     - name: app-user
+      credsSecret:
+        name: rustfs-user-app-user
       policies:
         - app-readwrite
   buckets:
@@ -157,7 +159,7 @@ spec:
       objectLock: true
 ```
 
-Policy ConfigMaps and user Secrets must live in the Tenant namespace. If they are created outside the Operator Console, add `rustfs.tenant=<tenant-name>` so changes to those resources enqueue the owning Tenant. Provisioned resources are retained when removed from the Tenant spec.
+Policy ConfigMaps and user Secrets must live in the Tenant namespace. `users[].credsSecret.name` selects the credentials Secret; when omitted, the operator falls back to a Secret named after `users[].name` for compatibility with existing manifests. The operator maintains `rustfs.tenant=<tenant-name>` on referenced policy ConfigMaps and user Secrets so changes enqueue the owning Tenant; references that do not exist yet are retried. The label triggers reconciliation but does not select the Secret. Provisioned resources are retained when removed from the Tenant spec.
 
 ### RBAC Configuration
 
@@ -333,9 +335,13 @@ kubectl apply -f examples/simple-tenant.yaml
 To upgrade the operator:
 
 ```bash
+kubectl apply -f deploy/rustfs-operator/crds/tenant-crd.yaml
 helm upgrade rustfs-operator deploy/rustfs-operator/
 ```
 
+Helm does not upgrade CRDs from the chart's `crds/` directory. Apply the updated Tenant CRD and wait for the operator rollout before using newly introduced Tenant fields such as `users[].credsSecret`.
+Existing manifests that omit the field remain compatible. Older operator binaries ignore the reference and continue using the same-name Secret convention, so complete the rollout before relying on it.
+
 ## Console UI
 
 The published `rustfs/operator` image contains both the Console backend (Rust API,
diff --git a/deploy/rustfs-operator/crds/tenant-crd.yaml b/deploy/rustfs-operator/crds/tenant-crd.yaml
index ea495a1..0a1bc1f 100644
--- a/deploy/rustfs-operator/crds/tenant-crd.yaml
+++ b/deploy/rustfs-operator/crds/tenant-crd.yaml
@@ -1761,6 +1761,17 @@ spec:
                 description: Regular users that should exist in the RustFS tenant.
                 items:
                   properties:
+                    credsSecret:
+                      description: Optional credentials Secret reference. Defaults to a Secret named after the user.
+                      nullable: true
+                      properties:
+                        name:
+                          maxLength: 253
+                          minLength: 1
+                          type: string
+                      required:
+                      - name
+                      type: object
                     deletionPolicy:
                       enum:
                       - Retain
@@ -1792,6 +1803,8 @@ spec:
                 x-kubernetes-validations:
                 - message: user policies must contain at least one policy
                   rule: self.all(x, has(x.policies) && x.policies.size() > 0)
+                - message: user credential Secret references must be unique
+                  rule: 'self.all(x, self.exists_one(y, (has(x.credsSecret) ? x.credsSecret.name : x.name) == (has(y.credsSecret) ? y.credsSecret.name : y.name)))'
             required:
             - pools
             type: object
@@ -2207,6 +2220,9 @@ spec:
                         objectLock:
                           nullable: true
                           type: boolean
+                        observedSecretName:
+                          nullable: true
+                          type: string
                         observedSecretResourceVersion:
                           nullable: true
                           type: string
@@ -2266,6 +2282,9 @@ spec:
                         objectLock:
                           nullable: true
                           type: boolean
+                        observedSecretName:
+                          nullable: true
+                          type: string
                         observedSecretResourceVersion:
                           nullable: true
                           type: string
@@ -2313,6 +2332,9 @@ spec:
                         objectLock:
                           nullable: true
                           type: boolean
+                        observedSecretName:
+                          nullable: true
+                          type: string
                         observedSecretResourceVersion:
                           nullable: true
                           type: string
diff --git a/deploy/rustfs-operator/crds/tenant.yaml b/deploy/rustfs-operator/crds/tenant.yaml
index ea495a1..0a1bc1f 100755
--- a/deploy/rustfs-operator/crds/tenant.yaml
+++ b/deploy/rustfs-operator/crds/tenant.yaml
@@ -1761,6 +1761,17 @@ spec:
                 description: Regular users that should exist in the RustFS tenant.
                 items:
                   properties:
+                    credsSecret:
+                      description: Optional credentials Secret reference. Defaults to a Secret named after the user.
+                      nullable: true
+                      properties:
+                        name:
+                          maxLength: 253
+                          minLength: 1
+                          type: string
+                      required:
+                      - name
+                      type: object
                     deletionPolicy:
                       enum:
                       - Retain
@@ -1792,6 +1803,8 @@ spec:
                 x-kubernetes-validations:
                 - message: user policies must contain at least one policy
                   rule: self.all(x, has(x.policies) && x.policies.size() > 0)
+                - message: user credential Secret references must be unique
+                  rule: 'self.all(x, self.exists_one(y, (has(x.credsSecret) ? x.credsSecret.name : x.name) == (has(y.credsSecret) ? y.credsSecret.name : y.name)))'
             required:
             - pools
             type: object
@@ -2207,6 +2220,9 @@ spec:
                         objectLock:
                           nullable: true
                           type: boolean
+                        observedSecretName:
+                          nullable: true
+                          type: string
                         observedSecretResourceVersion:
                           nullable: true
                           type: string
@@ -2266,6 +2282,9 @@ spec:
                         objectLock:
                           nullable: true
                           type: boolean
+                        observedSecretName:
+                          nullable: true
+                          type: string
                         observedSecretResourceVersion:
                           nullable: true
                           type: string
@@ -2313,6 +2332,9 @@ spec:
                         objectLock:
                           nullable: true
                           type: boolean
+                        observedSecretName:
+                          nullable: true
+                          type: string
                         observedSecretResourceVersion:
                           nullable: true
                           type: string
diff --git a/docs/operator-user-guide.md b/docs/operator-user-guide.md
index 4bd9b00..8d3bc3b 100644
--- a/docs/operator-user-guide.md
+++ b/docs/operator-user-guide.md
@@ -90,10 +90,14 @@ kubectl logs -n rustfs-system \
 Upgrade an existing installation:
 
 ```bash
+kubectl apply -f deploy/rustfs-operator/crds/tenant-crd.yaml
 helm upgrade rustfs-operator deploy/rustfs-operator/ \
   --namespace rustfs-system
 ```
 
+Helm does not upgrade CRDs from a chart's `crds/` directory. Apply the updated Tenant CRD before using fields introduced by a newer operator version, then wait for the operator rollout to complete.
+Existing Tenant manifests that omit newer optional fields require no migration. Do not rely on `users[].credsSecret` while any older operator binary is still reconciling the Tenant, because older versions use the same-name Secret convention.
+
 Uninstall:
 
 ```bash
@@ -594,13 +598,15 @@ The operator can create RustFS policies, users, and buckets after the Tenant wor
 - `spec.users` for regular users. Each user must have at least one direct policy mapping.
 - `spec.buckets` for buckets and optional object lock.
 
-ConfigMaps and user Secrets must live in the Tenant namespace. If managed outside the Operator Console, label them with `rustfs.tenant=<tenant-name>` so updates enqueue the owning Tenant.
+ConfigMaps and user Secrets must live in the Tenant namespace. The operator maintains `rustfs.tenant=<tenant-name>` on referenced policy ConfigMaps and user Secrets so updates enqueue the owning Tenant. References that do not exist yet are retried after they are created.
 
 Policy documents are parsed by RustFS. Use S3 ARN resource patterns such as `arn:aws:s3:::bucket` and `arn:aws:s3:::bucket/*`; for all buckets, use `arn:aws:s3:::*`. A bare `Resource: "*"` is not accepted by RustFS policy parsing.
 
-For each `spec.users[]` entry, the operator reads a Secret with the same name as the user. The Secret must contain `accesskey` and `secretkey`, or the MinIO-compatible keys `CONSOLE_ACCESS_KEY` and `CONSOLE_SECRET_KEY`. If both key formats are present, their values must match. User access keys must be at least 8 characters and must not contain whitespace, `=`, or `,`; user secret keys must be at least 8 characters.
+For each `spec.users[]` entry, set `credsSecret.name` to select a user credentials Secret in the Tenant namespace. If `credsSecret` is omitted, the operator reads a Secret with the same name as `user.name`, preserving the behavior of older manifests. An explicit reference is authoritative and does not fall back to the same-name Secret when it is invalid or missing. Each resolved Secret name must be unique within the Tenant; the API rejects duplicate references, and reconciliation also blocks them when an older installed CRD does not enforce the validation. Distinct Secrets must also contain distinct `accesskey` values. Reconciliation validates all user credentials before modifying any RustFS user and rejects every colliding entry.
+
+The Secret must contain `accesskey` and `secretkey`, or the MinIO-compatible keys `CONSOLE_ACCESS_KEY` and `CONSOLE_SECRET_KEY`. If both key formats are present, their values must match. `user.name` remains the declarative and status identity; the Secret's `accesskey` is the actual RustFS user. User access keys must be at least 8 characters and must not contain whitespace, `=`, or `,`; user secret keys must be at least 8 characters. The `rustfs.tenant` label only causes Secret updates to enqueue the Tenant; it does not select which Secret is read.
 
-Updating a user Secret's `secretkey` rotates that RustFS user's credential. The `accesskey` is immutable after the first successful reconciliation; use a new user entry and Secret when it must change, then migrate clients before removing the old entry.
+Updating a user Secret's `secretkey`, or changing `credsSecret.name` to another Secret with the same `accesskey`, rotates that RustFS user's credential. The `accesskey` is immutable after the first successful reconciliation; use a new user entry and Secret when it must change, then migrate clients before removing the old entry.
 
 ```yaml
 apiVersion: v1
@@ -626,7 +632,7 @@ data:
 apiVersion: v1
 kind: Secret
 metadata:
-  name: app-user
+  name: rustfs-user-app-user
   namespace: storage
   labels:
     rustfs.tenant: rustfs-a
@@ -656,6 +662,8 @@ spec:
           key: policy.json
   users:
     - name: app-user
+      credsSecret:
+        name: rustfs-user-app-user
       policies:
         - app-readwrite
   buckets:
diff --git a/docs/operator-user-guide.zh-CN.md b/docs/operator-user-guide.zh-CN.md
index c65408a..55cd34e 100644
--- a/docs/operator-user-guide.zh-CN.md
+++ b/docs/operator-user-guide.zh-CN.md
@@ -92,10 +92,14 @@ kubectl logs -n rustfs-system \
 升级已有安装：
 
 ```bash
+kubectl apply -f deploy/rustfs-operator/crds/tenant-crd.yaml
 helm upgrade rustfs-operator deploy/rustfs-operator/ \
   --namespace rustfs-system
 ```
 
+Helm 不会升级 Chart `crds/` 目录中的已有 CRD。使用新版本 Operator 引入的字段前，请先应用更新后的 Tenant CRD，并等待 Operator rollout 完成。
+未配置新版可选字段的已有 Tenant manifest 无需迁移。如果仍有旧版 Operator binary 在 reconcile Tenant，请勿依赖 `users[].credsSecret`，因为旧版本仍会按 user 同名规则读取 Secret。
+
 卸载：
 
 ```bash
@@ -594,13 +598,15 @@ Operator 可以在 Tenant workload Ready 后自动创建 RustFS policy、user 
 - `spec.users`：普通用户。每个 user 必须至少直接绑定一个 policy。
 - `spec.buckets`：bucket，可选择开启 object lock。
 
-ConfigMap 和 user Secret 必须位于 Tenant namespace。若这些资源不是通过 Operator Console 创建，建议添加 label：`rustfs.tenant=<tenant-name>`，这样资源变化可以触发 owning Tenant reconcile。
+ConfigMap 和 user Secret 必须位于 Tenant namespace。operator 会在被引用的 policy ConfigMap 和 user Secret 上维护 `rustfs.tenant=<tenant-name>` label，使资源变化能够触发 owning Tenant reconcile；尚未创建的引用资源会在创建后重试加 label。
 
 Policy document 由 RustFS 解析。请使用 `arn:aws:s3:::bucket` 和 `arn:aws:s3:::bucket/*` 这类 S3 ARN resource 写法；如需匹配所有 bucket，请使用 `arn:aws:s3:::*`。RustFS policy parser 不接受裸 `Resource: "*"`。
 
-每个 `spec.users[]` 条目都会读取一个与 user 名同名的 Secret。Secret 必须包含 `accesskey` 和 `secretkey`，或者 MinIO 兼容 key：`CONSOLE_ACCESS_KEY` 和 `CONSOLE_SECRET_KEY`。如果两种 key 同时存在，值必须一致。user access key 至少 8 个字符，且不能包含空白、`=` 或 `,`；user secret key 至少 8 个字符。
+每个 `spec.users[]` 条目都可以通过 `credsSecret.name` 指定 Tenant namespace 中的 user credentials Secret。省略 `credsSecret` 时，Operator 继续读取与 `user.name` 同名的 Secret，以兼容旧版 manifest。显式引用是唯一来源；配置错误或 Secret 不存在时，不会再回退到同名 Secret。同一 Tenant 内解析后的 Secret 名称必须唯一；API 会拒绝重复引用，而在集群仍安装旧版 CRD、尚未启用该校验时，reconcile 也会阻止这些重复配置生效。不同 Secret 中的 `accesskey` 也必须唯一；reconcile 会先校验全部 user credentials，再修改任何 RustFS user，并拒绝所有冲突条目。
+
+Secret 必须包含 `accesskey` 和 `secretkey`，或者 MinIO 兼容 key：`CONSOLE_ACCESS_KEY` 和 `CONSOLE_SECRET_KEY`。如果两种 key 同时存在，值必须一致。`user.name` 仍是声明和 status 中的逻辑标识，Secret 内的 `accesskey` 才是实际 RustFS user。user access key 至少 8 个字符，且不能包含空白、`=` 或 `,`；user secret key 至少 8 个字符。`rustfs.tenant` label 只负责在 Secret 更新时触发 Tenant reconcile，不参与选择要读取的 Secret。
 
-更新 user Secret 的 `secretkey` 会轮换对应 RustFS user 的凭据。首次成功 provisioning 后，`accesskey` 不可变；如需变更，请新建 user 条目和 Secret，迁移客户端后再移除旧条目。
+更新 user Secret 的 `secretkey`，或者把 `credsSecret.name` 切换到具有相同 `accesskey` 的另一个 Secret，都会轮换对应 RustFS user 的凭据。首次成功 provisioning 后，`accesskey` 不可变；如需变更，请新建 user 条目和 Secret，迁移客户端后再移除旧条目。
 
 ```yaml
 apiVersion: v1
@@ -626,7 +632,7 @@ data:
 apiVersion: v1
 kind: Secret
 metadata:
-  name: app-user
+  name: rustfs-user-app-user
   namespace: storage
   labels:
     rustfs.tenant: rustfs-a
@@ -656,6 +662,8 @@ spec:
           key: policy.json
   users:
     - name: app-user
+      credsSecret:
+        name: rustfs-user-app-user
       policies:
         - app-readwrite
   buckets:
diff --git a/examples/README.md b/examples/README.md
index 9d5aecc..a1c1790 100755
--- a/examples/README.md
+++ b/examples/README.md
@@ -130,17 +130,20 @@ Demonstrates operator-managed RustFS canned policies, regular users, and buckets
 **Features demonstrated:**
 - Tenant admin credentials through `spec.credsSecret`
 - Policy document stored in a labeled ConfigMap
-- User credentials stored in a labeled Secret
+- User credentials stored in a labeled Secret selected by `users[].credsSecret.name`
 - Required non-empty direct policy mapping for each user
 - Bucket creation with object lock verification
 
+For backward compatibility, omitting `users[].credsSecret` still selects a Secret
+with the same name as the logical user.
+
 **Deployment:**
 ```bash
 kubectl apply -f examples/provisioning-tenant.yaml
 kubectl wait --for=condition=Ready tenant/provisioning-demo --timeout=10m
 ```
 
-ConfigMaps and user Secrets should carry `rustfs.tenant=<tenant-name>` when they are managed outside the Console so updates enqueue the owning Tenant.
+The operator maintains `rustfs.tenant=<tenant-name>` on referenced policy ConfigMaps and user Secrets so updates enqueue the owning Tenant.
 
 ---
 
diff --git a/examples/provisioning-tenant.yaml b/examples/provisioning-tenant.yaml
index 59b0e1f..99bc3c9 100644
--- a/examples/provisioning-tenant.yaml
+++ b/examples/provisioning-tenant.yaml
@@ -47,7 +47,7 @@ data:
 apiVersion: v1
 kind: Secret
 metadata:
-  name: provisioning-app-user
+  name: rustfs-user-app-user
   namespace: default
   labels:
     rustfs.tenant: provisioning-demo
@@ -80,7 +80,9 @@ spec:
           key: policy.json
 
   users:
-    - name: provisioning-app-user
+    - name: app-user
+      credsSecret:
+        name: rustfs-user-app-user
       policies:
         - app-readwrite
 
diff --git a/src/console/handlers/tenants.rs b/src/console/handlers/tenants.rs
index d0f0848..27753aa 100755
--- a/src/console/handlers/tenants.rs
+++ b/src/console/handlers/tenants.rs
@@ -21,18 +21,14 @@ use crate::types::v1alpha1::{
     encryption::PodSecurityContextOverride,
     persistence::PersistenceConfig,
     pool::{Pool, validate_pool_shape_immutable},
-    tenant::{Tenant, TenantSpec},
+    tenant::Tenant,
 };
 use axum::{
     Extension, Json,
     extract::{Path, Query},
 };
 use k8s_openapi::api::core::v1 as corev1;
-use kube::{
-    Api, Client, ResourceExt,
-    api::{ListParams, Patch, PatchParams},
-};
-use serde_json::json;
+use kube::{Api, Client, ResourceExt, api::ListParams};
 
 // curl -s -X POST http://localhost:9090/api/v1/login \
 //   -H "Content-Type: application/json" \
@@ -310,14 +306,12 @@ pub async fn create_tenant(
         });
     }
 
-    let api: Api<Tenant> = Api::namespaced(client.clone(), &req.namespace);
+    let api: Api<Tenant> = Api::namespaced(client, &req.namespace);
     let created = api
         .create(&Default::default(), &tenant)
         .await
         .map_err(|e| error::map_kube_error(e, format!("Tenant '{}'", req.name)))?;
 
-    label_provisioning_references(&client, &req.namespace, &req.name, &created.spec).await;
-
     let item = tenant_to_list_item(created);
 
     Ok(Json(item))
@@ -348,14 +342,13 @@ pub async fn update_tenant(
     Json(req): Json<UpdateTenantRequest>,
 ) -> Result<Json<UpdateTenantResponse>> {
     let client = create_client(&claims).await?;
-    let api: Api<Tenant> = Api::namespaced(client.clone(), &namespace);
+    let api: Api<Tenant> = Api::namespaced(client, &namespace);
 
     // Load current object
     let mut tenant = api
         .get(&name)
         .await
         .map_err(|e| error::map_kube_error(e, format!("Tenant '{}'", name)))?;
-
     // Merge only provided fields
     let mut updated_fields = Vec::new();
 
@@ -486,8 +479,6 @@ pub async fn update_tenant(
         .await
         .map_err(|e| error::map_kube_error(e, format!("Tenant '{}'", name)))?;
 
-    label_provisioning_references(&client, &namespace, &name, &updated_tenant.spec).await;
-
     Ok(Json(UpdateTenantResponse {
         success: true,
         message: format!("Tenant updated: {}", updated_fields.join(", ")),
@@ -563,7 +554,6 @@ pub async fn put_tenant_yaml(
         .get(&name)
         .await
         .map_err(|e| error::map_kube_error(e, format!("Tenant '{}'", name)))?;
-
     if let Err(message) = validate_pool_shape_immutable(&current.spec.pools, &in_tenant.spec.pools)
     {
         return Err(Error::BadRequest { message });
@@ -660,58 +650,6 @@ fn summarize_tenant_states(tenants: &[Tenant]) -> TenantStateCountsResponse {
     }
 }
 
-async fn label_provisioning_references(
-    client: &Client,
-    namespace: &str,
-    tenant_name: &str,
-    spec: &TenantSpec,
-) {
-    let patch = json!({
-        "metadata": {
-            "labels": {
-                "rustfs.tenant": tenant_name,
-            },
-        },
-    });
-    let params = PatchParams::default();
-
-    let config_maps: std::collections::BTreeSet<_> = spec
-        .policies
-        .iter()
-        .map(|policy| policy.document.config_map_key_ref.name.as_str())
-        .collect();
-    let config_map_api: Api<corev1::ConfigMap> = Api::namespaced(client.clone(), namespace);
-    for name in config_maps {
-        if let Err(error) = config_map_api
-            .patch(name, &params, &Patch::Merge(&patch))
-            .await
-        {
-            tracing::debug!(
-                namespace,
-                tenant = tenant_name,
-                config_map = name,
-                %error,
-                "Failed to label provisioning policy ConfigMap"
-            );
-        }
-    }
-
-    let secrets: std::collections::BTreeSet<_> =
-        spec.users.iter().map(|user| user.name.as_str()).collect();
-    let secret_api: Api<corev1::Secret> = Api::namespaced(client.clone(), namespace);
-    for name in secrets {
-        if let Err(error) = secret_api.patch(name, &params, &Patch::Merge(&patch)).await {
-            tracing::debug!(
-                namespace,
-                tenant = tenant_name,
-                secret = name,
-                %error,
-                "Failed to label provisioning user Secret"
-            );
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::state_matches_filter;
diff --git a/src/console/openapi.rs b/src/console/openapi.rs
index baca60b..7f78338 100644
--- a/src/console/openapi.rs
+++ b/src/console/openapi.rs
@@ -50,7 +50,7 @@ use crate::console::models::topology::{
 };
 use crate::types::v1alpha1::provisioning::{
     ConfigMapKeyReference, PolicyDocumentSource, ProvisioningBucket, ProvisioningDeletionPolicy,
-    ProvisioningPolicy, ProvisioningUser,
+    ProvisioningPolicy, ProvisioningUser, UserCredentialsSecretRef,
 };
 use crate::types::v1alpha1::status::provisioning::{
     ProvisioningItemState, ProvisioningItemStatus, ProvisioningPhase, ProvisioningStatus,
@@ -109,6 +109,7 @@ use crate::types::v1alpha1::status::provisioning::{
         ProvisioningItemState,
         ProvisioningPolicy,
         ProvisioningUser,
+        UserCredentialsSecretRef,
         ProvisioningBucket,
         ProvisioningDeletionPolicy,
         PolicyDocumentSource,
@@ -434,6 +435,7 @@ mod tests {
             .expect("schemas exist");
 
         assert!(schemas.contains_key("ProvisioningStatus"));
+        assert!(schemas.contains_key("UserCredentialsSecretRef"));
         assert_eq!(
             spec.pointer("/components/schemas/TenantDetailsResponse/properties/provisioning/$ref")
                 .and_then(Value::as_str),
@@ -449,5 +451,12 @@ mod tests {
                 .and_then(Value::as_str),
             Some("#/components/schemas/ProvisioningBucket")
         );
+        assert_eq!(
+            spec.pointer(
+                "/components/schemas/ProvisioningUser/properties/credsSecret/oneOf/1/$ref",
+            )
+            .and_then(Value::as_str),
+            Some("#/components/schemas/UserCredentialsSecretRef")
+        );
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 811cee4..1c113ec 100755
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -17,7 +17,7 @@
 use crate::context::Context;
 use crate::reconcile::{error_policy, reconcile_rustfs};
 use crate::types::v1alpha1::policy_binding::PolicyBinding;
-use crate::types::v1alpha1::tenant::Tenant;
+use crate::types::v1alpha1::tenant::{RUSTFS_TENANT_LABEL, Tenant};
 use axum::{
     Router, body::Body, extract::State, http::StatusCode, middleware, response::IntoResponse,
     routing::get,
@@ -48,7 +48,6 @@ use tokio_util::sync::CancellationToken;
 use tower::ServiceExt as _;
 use tracing::{info, warn};
 
-const RUSTFS_TENANT_LABEL: &str = "rustfs.tenant";
 const CERT_MANAGER_GROUP: &str = "cert-manager.io";
 const CERT_MANAGER_VERSION: &str = "v1";
 const CERT_MANAGER_CERTIFICATE_KIND: &str = "Certificate";
diff --git a/src/reconcile.rs b/src/reconcile.rs
index 012cba8..9887a93 100755
--- a/src/reconcile.rs
+++ b/src/reconcile.rs
@@ -32,6 +32,7 @@ use tracing::{debug, info, warn};
 mod phases;
 mod pool_lifecycle;
 mod provisioning;
+mod reference_labels;
 mod tls;
 
 const OUT_OF_SERVICE_TAINT_KEY: &str = "node.kubernetes.io/out-of-service";
@@ -42,6 +43,9 @@ use phases::{
     reconcile_services, validate_no_pool_rename, validate_tenant_prerequisites,
 };
 use pool_lifecycle::reconcile_pool_lifecycle;
+use reference_labels::{
+    MISSING_REFERENCE_REQUEUE_INTERVAL, reconcile_provisioning_reference_labels,
+};
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -107,7 +111,13 @@ pub async fn reconcile_rustfs(tenant: Arc<Tenant>, ctx: Arc<Context>) -> Result<
         &removed_pool_cleanup,
     )
     .await?;
-    finalize_tenant_status(&ctx, &latest_tenant, summary, tls_plan).await
+    let action = finalize_tenant_status(&ctx, &latest_tenant, summary, tls_plan).await?;
+    let reference_labels =
+        reconcile_provisioning_reference_labels(&ctx, &latest_tenant, &ns).await?;
+    if reference_labels.has_missing_resources && action == Action::await_change() {
+        return Ok(Action::requeue(MISSING_REFERENCE_REQUEUE_INTERVAL));
+    }
+    Ok(action)
 }
 
 #[cfg(test)]
@@ -342,7 +352,7 @@ fn condition_marker(
         .map(|condition| (condition.status.clone(), condition.reason.clone()))
 }
 
-fn object_owned_by_tenant(metadata: &metav1::ObjectMeta, tenant: &Tenant) -> bool {
+pub(super) fn object_owned_by_tenant(metadata: &metav1::ObjectMeta, tenant: &Tenant) -> bool {
     let Some(tenant_uid) = tenant.metadata.uid.as_deref().filter(|uid| !uid.is_empty()) else {
         return false;
     };
diff --git a/src/reconcile/provisioning.rs b/src/reconcile/provisioning.rs
index 5254b2f..5897f49 100644
--- a/src/reconcile/provisioning.rs
+++ b/src/reconcile/provisioning.rs
@@ -16,6 +16,7 @@ use crate::context::{self, Context};
 use crate::sts::rustfs_client::{CreateBucketResult, RustfsAdminClient, RustfsClientError};
 use crate::types::v1alpha1::provisioning::{
     ProvisioningBucket, ProvisioningPolicy, ProvisioningUser,
+    duplicate_user_credentials_secret_names,
 };
 use crate::types::v1alpha1::status::Reason;
 use crate::types::v1alpha1::status::provisioning::{
@@ -53,9 +54,23 @@ struct ProvisioningRun<'a> {
 struct UserCredentials {
     access_key: String,
     secret_key: String,
+    secret_name: String,
     resource_version: Option<String>,
 }
 
+enum UserCredentialsCheck {
+    DuplicateSecret,
+    Checked {
+        policy_error: Option<String>,
+        credentials: Result<UserCredentials, String>,
+    },
+}
+
+struct UserCredentialsPreflight {
+    checks: Vec<UserCredentialsCheck>,
+    duplicate_access_key_hashes: BTreeSet<String>,
+}
+
 struct PolicyDocument {
     raw: String,
     normalized: String,
@@ -201,6 +216,7 @@ impl ProvisioningRun<'_> {
         item.last_applied_hash = previous.last_applied_hash.clone();
         item.last_applied_generation = previous.last_applied_generation;
         item.observed_secret_resource_version = previous.observed_secret_resource_version.clone();
+        item.observed_secret_name = previous.observed_secret_name.clone();
         item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone();
         item.policies = previous.policies.clone();
         item.region = previous.region.clone();
@@ -235,6 +251,7 @@ impl ProvisioningRun<'_> {
             if let Some(previous) = self.previous_user(&user.name) {
                 item.observed_secret_resource_version =
                     previous.observed_secret_resource_version.clone();
+                item.observed_secret_name = previous.observed_secret_name.clone();
                 item.last_applied_access_key_hash = previous.last_applied_access_key_hash.clone();
                 item.policies = previous.policies.clone();
             }
@@ -375,6 +392,7 @@ pub(super) async fn reconcile_provisioning(
             };
         }
     };
+    let user_credentials = preflight_user_credentials(&run).await;
 
     let mut live_policies = match load_live_policies(&client, tenant).await {
         Ok(policies) => policies,
@@ -399,7 +417,7 @@ pub(super) async fn reconcile_provisioning(
     };
 
     reconcile_policies(&mut run, &client, &mut live_policies).await;
-    reconcile_users(&mut run, &client, &live_policies).await;
+    reconcile_users(&mut run, &client, &live_policies, &user_credentials).await;
     reconcile_buckets(&mut run, &client).await;
     run.finish()
 }
@@ -692,6 +710,7 @@ async fn reconcile_users(
     run: &mut ProvisioningRun<'_>,
     client: &RustfsAdminClient,
     live_policies: &BTreeMap<String, String>,
+    credentials_preflight: &UserCredentialsPreflight,
 ) {
     let failed_spec_policies = run
         .status
@@ -701,46 +720,152 @@ async fn reconcile_users(
         .map(|item| item.name.clone())
         .collect::<BTreeSet<_>>();
 
-    for user in &run.tenant.spec.users {
-        let item = reconcile_user(run, client, live_policies, &failed_spec_policies, user).await;
+    for (user, preflight) in run
+        .tenant
+        .spec
+        .users
+        .iter()
+        .zip(credentials_preflight.checks.iter())
+    {
+        let (policy_error, credentials) = match preflight {
+            UserCredentialsCheck::DuplicateSecret => {
+                let previous = run.previous_user(&user.name);
+                let item = run.item(
+                    previous,
+                    &user.name,
+                    ProvisioningItemState::Failed,
+                    Reason::UserSecretInvalid,
+                    format!(
+                        "credentials Secret '{}' is referenced by multiple provisioning users",
+                        user.credentials_secret_name()
+                    ),
+                );
+                let item = annotate_user_item(item, user, previous, None);
+                run.push_user(item);
+                continue;
+            }
+            UserCredentialsCheck::Checked {
+                policy_error,
+                credentials,
+            } => (policy_error, credentials),
+        };
+        if let Ok(credentials) = credentials
+            && credentials_preflight
+                .duplicate_access_key_hashes
+                .contains(&access_key_hash(&credentials.access_key))
+        {
+            let previous = run.previous_user(&user.name);
+            let item = run.item(
+                previous,
+                &user.name,
+                ProvisioningItemState::Failed,
+                Reason::UserSecretInvalid,
+                format!(
+                    "credentials Secret '{}' resolves to an access key used by multiple provisioning users",
+                    credentials.secret_name
+                ),
+            );
+            let item = annotate_user_item(item, user, previous, None);
+            run.push_user(item);
+            continue;
+        }
+        if let Some(message) = policy_error {
+            let previous = run.previous_user(&user.name);
+            let item = run.item(
+                previous,
+                &user.name,
+                ProvisioningItemState::Failed,
+                Reason::UserPolicyInvalid,
+                message,
+            );
+            let item = annotate_user_item(item, user, previous, None);
+            run.push_user(item);
+            continue;
+        }
+        let credentials = match credentials {
+            Ok(credentials) => credentials,
+            Err(message) => {
+                let previous = run.previous_user(&user.name);
+                let item = run.item(
+                    previous,
+                    &user.name,
+                    ProvisioningItemState::Failed,
+                    Reason::UserSecretInvalid,
+                    message,
+                );
+                let item = annotate_user_item(item, user, previous, None);
+                run.push_user(item);
+                continue;
+            }
+        };
+
+        let item = reconcile_user(
+            run,
+            client,
+            live_policies,
+            &failed_spec_policies,
+            user,
+            credentials,
+        )
+        .await;
         run.push_user(item);
     }
 }
 
+async fn preflight_user_credentials(run: &ProvisioningRun<'_>) -> UserCredentialsPreflight {
+    let duplicate_secret_names = duplicate_user_credentials_secret_names(&run.tenant.spec.users);
+    let mut checks = Vec::with_capacity(run.tenant.spec.users.len());
+    for user in &run.tenant.spec.users {
+        if duplicate_secret_names.contains(user.credentials_secret_name()) {
+            checks.push(UserCredentialsCheck::DuplicateSecret);
+            continue;
+        }
+        checks.push(UserCredentialsCheck::Checked {
+            policy_error: validate_user_policies(user).err(),
+            credentials: load_user_secret(run, user).await,
+        });
+    }
+    let duplicate_access_key_hashes = duplicate_user_access_key_hashes(&checks);
+    UserCredentialsPreflight {
+        checks,
+        duplicate_access_key_hashes,
+    }
+}
+
+fn duplicate_user_access_key_hashes(
+    credentials_preflight: &[UserCredentialsCheck],
+) -> BTreeSet<String> {
+    let mut seen = BTreeSet::new();
+    credentials_preflight
+        .iter()
+        .filter_map(|preflight| match preflight {
+            UserCredentialsCheck::Checked {
+                credentials: Ok(credentials),
+                ..
+            } => Some(credentials),
+            UserCredentialsCheck::DuplicateSecret
+            | UserCredentialsCheck::Checked {
+                credentials: Err(_),
+                ..
+            } => None,
+        })
+        .filter_map(|credentials| {
+            let hash = access_key_hash(&credentials.access_key);
+            (!seen.insert(hash.clone())).then_some(hash)
+        })
+        .collect()
+}
+
 async fn reconcile_user(
     run: &ProvisioningRun<'_>,
     client: &RustfsAdminClient,
     live_policies: &BTreeMap<String, String>,
     failed_spec_policies: &BTreeSet<String>,
     user: &ProvisioningUser,
+    credentials: &UserCredentials,
 ) -> ProvisioningItemStatus {
     let previous = run.previous_user(&user.name);
-    if let Err(message) = validate_user_policies(user) {
-        let item = run.item(
-            previous,
-            &user.name,
-            ProvisioningItemState::Failed,
-            Reason::UserPolicyInvalid,
-            message,
-        );
-        return annotate_user_item(item, user, previous, None);
-    }
-
-    let credentials = match load_user_secret(run, user).await {
-        Ok(credentials) => credentials,
-        Err(message) => {
-            let item = run.item(
-                previous,
-                &user.name,
-                ProvisioningItemState::Failed,
-                Reason::UserSecretInvalid,
-                message,
-            );
-            return annotate_user_item(item, user, previous, None);
-        }
-    };
-
-    if user_access_key_changed(previous, &credentials) {
+    if user_access_key_changed(previous, credentials) {
         let item = run.item(
             previous,
             &user.name,
@@ -796,7 +921,7 @@ async fn reconcile_user(
     };
 
     let credentials_applied =
-        match sync_user_credentials(client, previous, &credentials, exists).await {
+        match sync_user_credentials(client, previous, credentials, exists).await {
             Ok(applied) => applied,
             Err(error) => {
                 let item = run.item(
@@ -821,7 +946,7 @@ async fn reconcile_user(
             Reason::UserPolicySetFailed,
             format!("failed to set RustFS user policy mapping: {error}"),
         );
-        return annotate_user_item(item, user, previous, Some(&credentials));
+        return annotate_user_item(item, user, previous, Some(credentials));
     }
 
     let message = if !exists {
@@ -847,7 +972,7 @@ async fn reconcile_user(
         }
         _ => Some(run.now.clone()),
     };
-    annotate_user_item(item, user, previous, Some(&credentials))
+    annotate_user_item(item, user, previous, Some(credentials))
 }
 
 fn annotate_user_item(
@@ -859,11 +984,13 @@ fn annotate_user_item(
     match applied_credentials {
         Some(credentials) => {
             item.observed_secret_resource_version = credentials.resource_version.clone();
+            item.observed_secret_name = Some(credentials.secret_name.clone());
             item.last_applied_access_key_hash = Some(access_key_hash(&credentials.access_key));
         }
         None => {
             item.observed_secret_resource_version =
                 previous.and_then(|item| item.observed_secret_resource_version.clone());
+            item.observed_secret_name = previous.and_then(|item| item.observed_secret_name.clone());
             item.last_applied_access_key_hash =
                 previous.and_then(|item| item.last_applied_access_key_hash.clone());
         }
@@ -916,6 +1043,7 @@ fn user_credentials_need_apply(
     let current_hash = access_key_hash(&credentials.access_key);
 
     previous.observed_secret_resource_version.as_deref() != Some(resource_version)
+        || previous.observed_secret_name.as_deref() != Some(credentials.secret_name.as_str())
         || previous.last_applied_access_key_hash.as_deref() != Some(current_hash.as_str())
 }
 
@@ -927,34 +1055,35 @@ async fn load_user_secret(
     run: &ProvisioningRun<'_>,
     user: &ProvisioningUser,
 ) -> Result<UserCredentials, String> {
+    let secret_name = user.credentials_secret_name();
     let secret: Secret = run
         .ctx
-        .get(&user.name, run.namespace)
+        .get(secret_name, run.namespace)
         .await
         .map_err(|error| {
             if context::is_kube_not_found(&error) {
-                format!("user Secret '{}' was not found", user.name)
+                format!("user Secret '{secret_name}' was not found")
             } else {
-                format!("failed to read user Secret '{}': {error}", user.name)
+                format!("failed to read user Secret '{secret_name}': {error}")
             }
         })?;
     let data = secret
         .data
         .as_ref()
-        .ok_or_else(|| format!("user Secret '{}' has no data", user.name))?;
+        .ok_or_else(|| format!("user Secret '{secret_name}' has no data"))?;
 
     let access_key = read_compatible_secret_value(
         data,
         "accesskey",
         "CONSOLE_ACCESS_KEY",
-        &user.name,
+        secret_name,
         "access key",
     )?;
     let secret_key = read_compatible_secret_value(
         data,
         "secretkey",
         "CONSOLE_SECRET_KEY",
-        &user.name,
+        secret_name,
         "secret key",
     )?;
 
@@ -964,6 +1093,7 @@ async fn load_user_secret(
     Ok(UserCredentials {
         access_key,
         secret_key,
+        secret_name: secret_name.to_string(),
         resource_version: secret.metadata.resource_version,
     })
 }
@@ -1355,11 +1485,18 @@ mod tests {
         body::Body,
         extract::State,
         http::{Request, StatusCode},
-        routing::{get, put},
+        routing::{any, get, put},
     };
     use k8s_openapi::ByteString;
-    use std::sync::Arc;
+    use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+    use kube::{Client, client::Body as KubeBody};
+    use std::convert::Infallible;
+    use std::sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    };
     use tokio::sync::Mutex;
+    use tower::service_fn;
 
     #[derive(Clone, Default)]
     struct PolicyApplyCapture {
@@ -1404,10 +1541,187 @@ mod tests {
         assert!(error.contains("at least 8 characters"));
     }
 
+    #[tokio::test]
+    async fn duplicate_access_keys_fail_before_any_rustfs_user_request() {
+        let kube_service = service_fn(move |request: http::Request<KubeBody>| async move {
+            let secret_name = request
+                .uri()
+                .path()
+                .rsplit('/')
+                .next()
+                .expect("Secret request should include a name");
+            let secret_key = match secret_name {
+                "user-secret-a" => b"secret-value-a".to_vec(),
+                "user-secret-b" => b"secret-value-b".to_vec(),
+                _ => panic!("unexpected Secret request: {secret_name}"),
+            };
+            let secret = Secret {
+                metadata: ObjectMeta {
+                    name: Some(secret_name.to_string()),
+                    namespace: Some("storage".to_string()),
+                    resource_version: Some("1".to_string()),
+                    ..Default::default()
+                },
+                data: Some(BTreeMap::from([
+                    (
+                        "accesskey".to_string(),
+                        ByteString(b"shareduser01".to_vec()),
+                    ),
+                    ("secretkey".to_string(), ByteString(secret_key)),
+                ])),
+                ..Default::default()
+            };
+            Ok::<_, Infallible>(
+                http::Response::builder()
+                    .body(KubeBody::from(
+                        serde_json::to_vec(&secret).expect("Secret should serialize"),
+                    ))
+                    .expect("response should build"),
+            )
+        });
+        let ctx = Context::new(Client::new(kube_service, "default"));
+        let mut colliding_with_invalid_policy =
+            provisioning_user("logical-b", "user-secret-b", "readonly");
+        colliding_with_invalid_policy.policies.clear();
+        let tenant = Tenant {
+            metadata: ObjectMeta {
+                name: Some("tenant-a".to_string()),
+                namespace: Some("storage".to_string()),
+                ..Default::default()
+            },
+            spec: crate::types::v1alpha1::tenant::TenantSpec {
+                users: vec![
+                    provisioning_user("logical-a", "user-secret-a", "readwrite"),
+                    colliding_with_invalid_policy,
+                ],
+                ..Default::default()
+            },
+            status: None,
+        };
+        let mut run = ProvisioningRun {
+            ctx: &ctx,
+            tenant: &tenant,
+            namespace: "storage",
+            previous: ProvisioningStatus::default(),
+            now: "2026-07-18T00:00:00Z".to_string(),
+            status: ProvisioningStatus::default(),
+            failures: Vec::new(),
+        };
+
+        let request_count = Arc::new(AtomicUsize::new(0));
+        let route_count = request_count.clone();
+        let router = Router::new().fallback(any(move || {
+            let route_count = route_count.clone();
+            async move {
+                route_count.fetch_add(1, Ordering::SeqCst);
+                StatusCode::OK
+            }
+        }));
+        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
+            .await
+            .expect("test server should bind");
+        let addr = listener.local_addr().expect("listener should have address");
+        let server = tokio::spawn(async move {
+            axum::serve(listener, router)
+                .await
+                .expect("test server should serve")
+        });
+        let client =
+            RustfsAdminClient::new_with_base_url(format!("http://{addr}"), "access", "secret");
+        let live_policies = BTreeMap::from([("readwrite".to_string(), "{}".to_string())]);
+
+        let credentials_preflight = preflight_user_credentials(&run).await;
+        reconcile_users(&mut run, &client, &live_policies, &credentials_preflight).await;
+
+        assert_eq!(request_count.load(Ordering::SeqCst), 0);
+        assert_eq!(run.status.users.len(), 2);
+        assert!(run.status.users.iter().all(|item| {
+            item.state == ProvisioningItemState::Failed.as_str()
+                && item.reason == Reason::UserSecretInvalid.as_str()
+                && item
+                    .message
+                    .as_deref()
+                    .is_some_and(|message| !message.contains("shareduser01"))
+        }));
+        server.abort();
+    }
+
+    #[tokio::test]
+    async fn user_preflight_loads_credentials_without_changing_policy_error_priority() {
+        let request_count = Arc::new(AtomicUsize::new(0));
+        let service_count = request_count.clone();
+        let kube_service = service_fn(move |_request: http::Request<KubeBody>| {
+            let service_count = service_count.clone();
+            async move {
+                service_count.fetch_add(1, Ordering::SeqCst);
+                Ok::<_, Infallible>(
+                    http::Response::builder()
+                        .status(StatusCode::INTERNAL_SERVER_ERROR)
+                        .body(KubeBody::empty())
+                        .expect("response should build"),
+                )
+            }
+        });
+        let ctx = Context::new(Client::new(kube_service, "default"));
+        let tenant = Tenant {
+            metadata: ObjectMeta {
+                name: Some("tenant-a".to_string()),
+                namespace: Some("storage".to_string()),
+                ..Default::default()
+            },
+            spec: crate::types::v1alpha1::tenant::TenantSpec {
+                users: vec![ProvisioningUser {
+                    name: "invalid-user".to_string(),
+                    creds_secret: Some(
+                        crate::types::v1alpha1::provisioning::UserCredentialsSecretRef {
+                            name: "missing-secret".to_string(),
+                        },
+                    ),
+                    policies: Vec::new(),
+                    deletion_policy: Default::default(),
+                }],
+                ..Default::default()
+            },
+            status: None,
+        };
+        let mut run = ProvisioningRun {
+            ctx: &ctx,
+            tenant: &tenant,
+            namespace: "storage",
+            previous: ProvisioningStatus::default(),
+            now: "2026-07-18T00:00:00Z".to_string(),
+            status: ProvisioningStatus::default(),
+            failures: Vec::new(),
+        };
+
+        let client = RustfsAdminClient::new_with_base_url(
+            "http://127.0.0.1:1".to_string(),
+            "access",
+            "secret",
+        );
+
+        let credentials_preflight = preflight_user_credentials(&run).await;
+        reconcile_users(&mut run, &client, &BTreeMap::new(), &credentials_preflight).await;
+
+        assert_eq!(request_count.load(Ordering::SeqCst), 1);
+        assert_eq!(run.status.users.len(), 1);
+        assert_eq!(
+            run.status.users[0].reason,
+            Reason::UserPolicyInvalid.as_str()
+        );
+        assert!(
+            run.status.users[0]
+                .message
+                .as_deref()
+                .is_some_and(|message| message.contains("at least one policy"))
+        );
+    }
+
     #[test]
     fn user_policy_list_must_not_be_empty() {
         let user = ProvisioningUser {
             name: "app-user".to_string(),
+            creds_secret: None,
             policies: Vec::new(),
             deletion_policy: Default::default(),
         };
@@ -1418,6 +1732,19 @@ mod tests {
         assert!(error.contains("at least one policy"));
     }
 
+    fn provisioning_user(name: &str, secret_name: &str, policy: &str) -> ProvisioningUser {
+        ProvisioningUser {
+            name: name.to_string(),
+            creds_secret: Some(
+                crate::types::v1alpha1::provisioning::UserCredentialsSecretRef {
+                    name: secret_name.to_string(),
+                },
+            ),
+            policies: vec![policy.to_string()],
+            deletion_policy: Default::default(),
+        }
+    }
+
     #[test]
     fn policy_document_hash_uses_compact_json() {
         let normalized = normalize_policy_document(
@@ -1599,10 +1926,12 @@ mod tests {
             Reason::ProvisioningConfigured.as_str(),
         );
         previous.observed_secret_resource_version = Some("1".to_string());
+        previous.observed_secret_name = Some("app-user".to_string());
         previous.last_applied_access_key_hash = Some(access_key_hash("appuser01"));
         let credentials = UserCredentials {
             access_key: "appuser01".to_string(),
             secret_key: "rotated-secret".to_string(),
+            secret_name: "app-user".to_string(),
             resource_version: Some("2".to_string()),
         };
 
@@ -1627,10 +1956,12 @@ mod tests {
             Reason::ProvisioningConfigured.as_str(),
         );
         previous.observed_secret_resource_version = Some("1".to_string());
+        previous.observed_secret_name = Some("app-user".to_string());
         previous.last_applied_access_key_hash = Some(access_key_hash("appuser01"));
         let credentials = UserCredentials {
             access_key: "appuser01".to_string(),
             secret_key: "unchanged-secret".to_string(),
+            secret_name: "app-user".to_string(),
             resource_version: Some("1".to_string()),
         };
 
@@ -1641,6 +1972,53 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn changed_user_secret_reference_requires_credentials_write() {
+        let mut previous = ProvisioningItemStatus::new(
+            "app-user",
+            ProvisioningItemState::Ready,
+            Reason::ProvisioningConfigured.as_str(),
+        );
+        previous.observed_secret_resource_version = Some("1".to_string());
+        previous.observed_secret_name = Some("app-user".to_string());
+        previous.last_applied_access_key_hash = Some(access_key_hash("appuser01"));
+        let credentials = UserCredentials {
+            access_key: "appuser01".to_string(),
+            secret_key: "rotated-secret".to_string(),
+            secret_name: "rustfs-user-app-user".to_string(),
+            resource_version: Some("1".to_string()),
+        };
+
+        assert!(user_credentials_need_apply(
+            Some(&previous),
+            &credentials,
+            true
+        ));
+    }
+
+    #[test]
+    fn legacy_user_status_records_secret_identity_with_one_credentials_write() {
+        let mut previous = ProvisioningItemStatus::new(
+            "app-user",
+            ProvisioningItemState::Ready,
+            Reason::ProvisioningConfigured.as_str(),
+        );
+        previous.observed_secret_resource_version = Some("1".to_string());
+        previous.last_applied_access_key_hash = Some(access_key_hash("appuser01"));
+        let credentials = UserCredentials {
+            access_key: "appuser01".to_string(),
+            secret_key: "unchanged-secret".to_string(),
+            secret_name: "app-user".to_string(),
+            resource_version: Some("1".to_string()),
+        };
+
+        assert!(user_credentials_need_apply(
+            Some(&previous),
+            &credentials,
+            true
+        ));
+    }
+
     #[test]
     fn changed_user_access_key_is_rejected() {
         let mut previous = ProvisioningItemStatus::new(
@@ -1652,6 +2030,7 @@ mod tests {
         let credentials = UserCredentials {
             access_key: "otheruser".to_string(),
             secret_key: "rotated-secret".to_string(),
+            secret_name: "app-user".to_string(),
             resource_version: Some("2".to_string()),
         };
 
diff --git a/src/reconcile/reference_labels.rs b/src/reconcile/reference_labels.rs
new file mode 100644
index 0000000..ac7b565
--- /dev/null
+++ b/src/reconcile/reference_labels.rs
@@ -0,0 +1,596 @@
+// Copyright 2025 RustFS Team
+//
+// 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.
+
+use super::{Error, object_owned_by_tenant};
+use crate::context::{self, Context};
+use crate::types::v1alpha1::tenant::{RUSTFS_TENANT_LABEL, Tenant};
+use k8s_openapi::NamespaceResourceScope;
+use k8s_openapi::api::core::v1::{ConfigMap, Secret};
+use kube::api::{ListParams, Patch, PatchParams};
+use kube::{Api, Resource, ResourceExt};
+use serde::de::DeserializeOwned;
+use serde_json::json;
+use std::collections::BTreeSet;
+use std::fmt::Debug;
+use std::time::Duration;
+use tracing::{debug, info};
+
+pub(super) const MISSING_REFERENCE_REQUEUE_INTERVAL: Duration = Duration::from_secs(60);
+
+#[derive(Default)]
+pub(super) struct ReferenceLabelReconcileResult {
+    pub(super) has_missing_resources: bool,
+}
+
+enum EnsureLabelOutcome {
+    Present,
+    Labeled,
+    Missing,
+}
+
+pub(super) async fn reconcile_provisioning_reference_labels(
+    ctx: &Context,
+    tenant: &Tenant,
+    namespace: &str,
+) -> Result<ReferenceLabelReconcileResult, Error> {
+    let tenant_name = tenant.name();
+    let desired_user_secrets = tenant
+        .spec
+        .users
+        .iter()
+        .map(|user| user.credentials_secret_name().to_string())
+        .collect::<BTreeSet<_>>();
+    let referenced_secrets = tenant.spec.referenced_secret_names();
+    let desired_policy_config_maps = tenant
+        .spec
+        .policies
+        .iter()
+        .map(|policy| policy.document.config_map_key_ref.name.clone())
+        .collect::<BTreeSet<_>>();
+
+    let secret_api = Api::<Secret>::namespaced(ctx.client.clone(), namespace);
+    let labeled_secrets = list_labeled_resources(&secret_api, &tenant_name).await?;
+    let mut result = ReferenceLabelReconcileResult {
+        has_missing_resources: ensure_resource_labels(
+            &secret_api,
+            "Secret",
+            namespace,
+            &tenant_name,
+            &desired_user_secrets,
+            &labeled_resource_names(&labeled_secrets),
+        )
+        .await?,
+    };
+    remove_stale_secret_labels(
+        &secret_api,
+        tenant,
+        namespace,
+        &referenced_secrets,
+        labeled_secrets,
+    )
+    .await?;
+
+    if !desired_policy_config_maps.is_empty() {
+        let config_map_api = Api::<ConfigMap>::namespaced(ctx.client.clone(), namespace);
+        let labeled_config_maps = list_labeled_resources(&config_map_api, &tenant_name).await?;
+        result.has_missing_resources |= ensure_resource_labels(
+            &config_map_api,
+            "ConfigMap",
+            namespace,
+            &tenant_name,
+            &desired_policy_config_maps,
+            &labeled_resource_names(&labeled_config_maps),
+        )
+        .await?;
+    }
+
+    Ok(result)
+}
+
+async fn list_labeled_resources<T>(api: &Api<T>, tenant_name: &str) -> Result<Vec<T>, Error>
+where
+    T: Clone + DeserializeOwned + Debug + Resource<Scope = NamespaceResourceScope>,
+    <T as Resource>::DynamicType: Default,
+{
+    let params = ListParams::default().labels(&format!("{RUSTFS_TENANT_LABEL}={tenant_name}"));
+    api.list(&params)
+        .await
+        .map(|resources| resources.items)
+        .map_err(|source| context::Error::Kube { source }.into())
+}
+
+fn labeled_resource_names<T>(resources: &[T]) -> BTreeSet<String>
+where
+    T: ResourceExt,
+{
+    resources.iter().map(ResourceExt::name_any).collect()
+}
+
+async fn ensure_resource_labels<T>(
+    api: &Api<T>,
+    resource_kind: &str,
+    namespace: &str,
+    tenant_name: &str,
+    desired_names: &BTreeSet<String>,
+    labeled_names: &BTreeSet<String>,
+) -> Result<bool, Error>
+where
+    T: Clone + DeserializeOwned + Debug + Resource<Scope = NamespaceResourceScope>,
+    <T as Resource>::DynamicType: Default,
+{
+    let mut has_missing_resources = false;
+    for name in desired_names.difference(labeled_names) {
+        match ensure_resource_label(api, name, tenant_name).await? {
+            EnsureLabelOutcome::Present => {}
+            EnsureLabelOutcome::Labeled => {
+                info!(
+                    namespace,
+                    tenant = tenant_name,
+                    resource_kind,
+                    resource = name,
+                    "labeled provisioning reference for Tenant watch"
+                );
+            }
+            EnsureLabelOutcome::Missing => {
+                has_missing_resources = true;
+                debug!(
+                    namespace,
+                    tenant = tenant_name,
+                    resource_kind,
+                    resource = name,
+                    "provisioning reference is not available for Tenant watch labeling"
+                );
+            }
+        }
+    }
+    Ok(has_missing_resources)
+}
+
+async fn ensure_resource_label<T>(
+    api: &Api<T>,
+    resource_name: &str,
+    tenant_name: &str,
+) -> Result<EnsureLabelOutcome, Error>
+where
+    T: Clone + DeserializeOwned + Debug + Resource<Scope = NamespaceResourceScope>,
+    <T as Resource>::DynamicType: Default,
+{
+    let resource = match api.get(resource_name).await {
+        Ok(resource) => resource,
+        Err(kube::Error::Api(response)) if response.code == 404 => {
+            return Ok(EnsureLabelOutcome::Missing);
+        }
+        Err(source) => return Err(context::Error::Kube { source }.into()),
+    };
+    if resource
+        .meta()
+        .labels
+        .as_ref()
+        .and_then(|labels| labels.get(RUSTFS_TENANT_LABEL))
+        .is_some_and(|owner| owner == tenant_name)
+    {
+        return Ok(EnsureLabelOutcome::Present);
+    }
+
+    let mut metadata = json!({
+        "labels": {
+            (RUSTFS_TENANT_LABEL): tenant_name,
+        },
+    });
+    if let Some(resource_version) = &resource.meta().resource_version {
+        metadata["resourceVersion"] = json!(resource_version);
+    }
+    let patch = json!({ "metadata": metadata });
+    api.patch(
+        resource_name,
+        &PatchParams::default(),
+        &Patch::Merge(&patch),
+    )
+    .await
+    .map_err(|source| context::Error::Kube { source })?;
+    Ok(EnsureLabelOutcome::Labeled)
+}
+
+async fn remove_stale_secret_labels(
+    secret_api: &Api<Secret>,
+    tenant: &Tenant,
+    namespace: &str,
+    referenced_secrets: &BTreeSet<String>,
+    labeled_secrets: Vec<Secret>,
+) -> Result<(), Error> {
+    for secret in labeled_secrets {
+        let name = secret.name_any();
+        if referenced_secrets.contains(&name) || object_owned_by_tenant(&secret.metadata, tenant) {
+            continue;
+        }
+
+        let mut metadata = json!({
+            "labels": {
+                (RUSTFS_TENANT_LABEL): null,
+            },
+        });
+        if let Some(resource_version) = &secret.metadata.resource_version {
+            metadata["resourceVersion"] = json!(resource_version);
+        }
+        let patch = json!({ "metadata": metadata });
+        secret_api
+            .patch(&name, &PatchParams::default(), &Patch::Merge(&patch))
+            .await
+            .map_err(|source| context::Error::Kube { source })?;
+        info!(
+            namespace,
+            tenant = %tenant.name(),
+            secret = name,
+            "removed stale Tenant watch label from provisioning Secret"
+        );
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::v1alpha1::provisioning::{
+        ConfigMapKeyReference, PolicyDocumentSource, ProvisioningPolicy, ProvisioningUser,
+        UserCredentialsSecretRef,
+    };
+    use crate::types::v1alpha1::tenant::{RpcSecretRef, TenantSpec};
+    use http::{Method, Request, Response, StatusCode};
+    use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, OwnerReference};
+    use kube::{Client, client::Body};
+    use serde_json::Value;
+    use std::convert::Infallible;
+    use std::sync::{Arc, Mutex};
+    use tower::service_fn;
+
+    #[derive(Debug)]
+    struct CapturedRequest {
+        method: Method,
+        path: String,
+        body: Option<Value>,
+    }
+
+    #[tokio::test]
+    async fn operator_labels_current_references_and_cleans_only_stale_external_secrets() {
+        let captured = Arc::new(Mutex::new(Vec::<CapturedRequest>::new()));
+        let service_capture = captured.clone();
+        let service = service_fn(move |request: Request<Body>| {
+            let service_capture = service_capture.clone();
+            async move {
+                let method = request.method().clone();
+                let path = request.uri().path().to_string();
+                let body = request
+                    .into_body()
+                    .collect_bytes()
+                    .await
+                    .expect("request body should be readable");
+                let body = (!body.is_empty()).then(|| {
+                    serde_json::from_slice(&body).expect("request body should be valid JSON")
+                });
+                service_capture
+                    .lock()
+                    .expect("request capture lock should be available")
+                    .push(CapturedRequest {
+                        method: method.clone(),
+                        path: path.clone(),
+                        body,
+                    });
+
+                let response = match (method.as_str(), path.as_str()) {
+                    ("GET", "/api/v1/namespaces/storage/secrets") => json_response(
+                        StatusCode::OK,
+                        json!({
+                            "apiVersion": "v1",
+                            "kind": "SecretList",
+                            "metadata": { "resourceVersion": "50" },
+                            "items": [
+                                labeled_secret("old-user-secret", "42", None),
+                                labeled_secret("shared-secret", "43", None),
+                                labeled_secret(
+                                    "owned-secret",
+                                    "44",
+                                    Some(tenant_owner_reference()),
+                                ),
+                            ],
+                        }),
+                    ),
+                    ("GET", "/api/v1/namespaces/storage/secrets/new-user-secret") => json_response(
+                        StatusCode::OK,
+                        json!({
+                            "apiVersion": "v1",
+                            "kind": "Secret",
+                            "metadata": {
+                                "name": "new-user-secret",
+                                "namespace": "storage",
+                                "resourceVersion": "7"
+                            }
+                        }),
+                    ),
+                    ("PATCH", "/api/v1/namespaces/storage/secrets/new-user-secret") => {
+                        json_response(StatusCode::OK, labeled_secret("new-user-secret", "8", None))
+                    }
+                    ("PATCH", "/api/v1/namespaces/storage/secrets/old-user-secret") => {
+                        json_response(
+                            StatusCode::OK,
+                            json!({
+                                "apiVersion": "v1",
+                                "kind": "Secret",
+                                "metadata": {
+                                    "name": "old-user-secret",
+                                    "namespace": "storage",
+                                    "resourceVersion": "43"
+                                }
+                            }),
+                        )
+                    }
+                    ("GET", "/api/v1/namespaces/storage/configmaps") => json_response(
+                        StatusCode::OK,
+                        json!({
+                            "apiVersion": "v1",
+                            "kind": "ConfigMapList",
+                            "metadata": { "resourceVersion": "51" },
+                            "items": [],
+                        }),
+                    ),
+                    ("GET", "/api/v1/namespaces/storage/configmaps/policy-config") => {
+                        json_response(
+                            StatusCode::OK,
+                            json!({
+                                "apiVersion": "v1",
+                                "kind": "ConfigMap",
+                                "metadata": {
+                                    "name": "policy-config",
+                                    "namespace": "storage",
+                                    "resourceVersion": "9"
+                                }
+                            }),
+                        )
+                    }
+                    ("PATCH", "/api/v1/namespaces/storage/configmaps/policy-config") => {
+                        json_response(
+                            StatusCode::OK,
+                            json!({
+                                "apiVersion": "v1",
+                                "kind": "ConfigMap",
+                                "metadata": {
+                                    "name": "policy-config",
+                                    "namespace": "storage",
+                                    "resourceVersion": "10",
+                                    "labels": { "rustfs.tenant": "tenant-a" }
+                                }
+                            }),
+                        )
+                    }
+                    _ => panic!("unexpected Kubernetes request: {method} {path}"),
+                };
+                Ok::<_, Infallible>(response)
+            }
+        });
+        let ctx = Context::new(Client::new(service, "default"));
+        let tenant = tenant_with_user_and_shared_rpc_secret();
+
+        let result = reconcile_provisioning_reference_labels(&ctx, &tenant, "storage")
+            .await
+            .expect("reference labels should reconcile");
+
+        assert!(!result.has_missing_resources);
+        let captured = captured
+            .lock()
+            .expect("request capture lock should be available");
+        let new_secret_patch = captured
+            .iter()
+            .find(|request| {
+                request.method == Method::PATCH && request.path.ends_with("/new-user-secret")
+            })
+            .expect("new user Secret should be labeled");
+        assert_eq!(
+            new_secret_patch.body,
+            Some(json!({
+                "metadata": {
+                    "resourceVersion": "7",
+                    "labels": { "rustfs.tenant": "tenant-a" }
+                }
+            }))
+        );
+        let stale_secret_patch = captured
+            .iter()
+            .find(|request| {
+                request.method == Method::PATCH && request.path.ends_with("/old-user-secret")
+            })
+            .expect("stale user Secret label should be removed");
+        assert_eq!(
+            stale_secret_patch.body,
+            Some(json!({
+                "metadata": {
+                    "resourceVersion": "42",
+                    "labels": { "rustfs.tenant": null }
+                }
+            }))
+        );
+        assert_eq!(
+            captured
+                .iter()
+                .filter(|request| request.method == Method::PATCH)
+                .count(),
+            3,
+            "shared and Tenant-owned Secrets must retain their labels"
+        );
+        let policy_config_map_patch = captured
+            .iter()
+            .find(|request| {
+                request.method == Method::PATCH && request.path.ends_with("/policy-config")
+            })
+            .expect("policy ConfigMap should be labeled");
+        assert_eq!(
+            policy_config_map_patch.body,
+            Some(json!({
+                "metadata": {
+                    "resourceVersion": "9",
+                    "labels": { "rustfs.tenant": "tenant-a" }
+                }
+            }))
+        );
+    }
+
+    #[tokio::test]
+    async fn missing_reference_requests_a_follow_up_without_failing_reconcile() {
+        let service = service_fn(move |request: Request<Body>| async move {
+            let method = request.method().clone();
+            let path = request.uri().path().to_string();
+            let response = match (method.as_str(), path.as_str()) {
+                ("GET", "/api/v1/namespaces/storage/secrets") => json_response(
+                    StatusCode::OK,
+                    json!({
+                        "apiVersion": "v1",
+                        "kind": "SecretList",
+                        "metadata": { "resourceVersion": "1" },
+                        "items": [],
+                    }),
+                ),
+                ("GET", "/api/v1/namespaces/storage/secrets/new-user-secret") => json_response(
+                    StatusCode::NOT_FOUND,
+                    json!({
+                        "apiVersion": "v1",
+                        "kind": "Status",
+                        "status": "Failure",
+                        "reason": "NotFound",
+                        "code": 404
+                    }),
+                ),
+                _ => panic!("unexpected Kubernetes request: {method} {path}"),
+            };
+            Ok::<_, Infallible>(response)
+        });
+        let ctx = Context::new(Client::new(service, "default"));
+        let mut tenant = tenant_with_user_and_shared_rpc_secret();
+        tenant.spec.rpc_secret = None;
+        tenant.spec.policies.clear();
+
+        let result = reconcile_provisioning_reference_labels(&ctx, &tenant, "storage")
+            .await
+            .expect("a missing provisioning reference is a pending condition");
+
+        assert!(result.has_missing_resources);
+    }
+
+    #[tokio::test]
+    async fn stale_label_conflict_is_returned_for_controller_retry() {
+        let service = service_fn(move |_request: Request<Body>| async move {
+            Ok::<_, Infallible>(json_response(
+                StatusCode::CONFLICT,
+                json!({
+                    "apiVersion": "v1",
+                    "kind": "Status",
+                    "status": "Failure",
+                    "reason": "Conflict",
+                    "code": 409
+                }),
+            ))
+        });
+        let secret_api = Api::<Secret>::namespaced(Client::new(service, "default"), "storage");
+        let tenant = tenant_with_user_and_shared_rpc_secret();
+        let stale_secret: Secret =
+            serde_json::from_value(labeled_secret("old-user-secret", "42", None))
+                .expect("Secret should deserialize");
+
+        let error = remove_stale_secret_labels(
+            &secret_api,
+            &tenant,
+            "storage",
+            &BTreeSet::new(),
+            vec![stale_secret],
+        )
+        .await
+        .expect_err("resourceVersion conflicts must be retried by the controller");
+
+        assert!(matches!(
+            error,
+            Error::Context {
+                source: context::Error::Kube {
+                    source: kube::Error::Api(response),
+                },
+            } if response.code == 409
+        ));
+    }
+
+    fn tenant_with_user_and_shared_rpc_secret() -> Tenant {
+        Tenant {
+            metadata: ObjectMeta {
+                name: Some("tenant-a".to_string()),
+                namespace: Some("storage".to_string()),
+                uid: Some("tenant-uid".to_string()),
+                ..Default::default()
+            },
+            spec: TenantSpec {
+                users: vec![ProvisioningUser {
+                    name: "app-user".to_string(),
+                    creds_secret: Some(UserCredentialsSecretRef {
+                        name: "new-user-secret".to_string(),
+                    }),
+                    ..Default::default()
+                }],
+                rpc_secret: Some(RpcSecretRef {
+                    name: "shared-secret".to_string(),
+                    key: "rpc-secret".to_string(),
+                }),
+                policies: vec![ProvisioningPolicy {
+                    name: "readwrite".to_string(),
+                    document: PolicyDocumentSource {
+                        config_map_key_ref: ConfigMapKeyReference {
+                            name: "policy-config".to_string(),
+                            key: "policy.json".to_string(),
+                        },
+                    },
+                    deletion_policy: Default::default(),
+                }],
+                ..Default::default()
+            },
+            status: None,
+        }
+    }
+
+    fn tenant_owner_reference() -> OwnerReference {
+        OwnerReference {
+            api_version: "rustfs.com/v1alpha1".to_string(),
+            kind: "Tenant".to_string(),
+            name: "tenant-a".to_string(),
+            uid: "tenant-uid".to_string(),
+            controller: Some(true),
+            block_owner_deletion: Some(true),
+        }
+    }
+
+    fn labeled_secret(name: &str, resource_version: &str, owner: Option<OwnerReference>) -> Value {
+        json!({
+            "apiVersion": "v1",
+            "kind": "Secret",
+            "metadata": {
+                "name": name,
+                "namespace": "storage",
+                "resourceVersion": resource_version,
+                "labels": { "rustfs.tenant": "tenant-a" },
+                "ownerReferences": owner.into_iter().collect::<Vec<_>>(),
+            }
+        })
+    }
+
+    fn json_response(status: StatusCode, value: Value) -> Response<Body> {
+        Response::builder()
+            .status(status)
+            .body(Body::from(
+                serde_json::to_vec(&value).expect("response should serialize"),
+            ))
+            .expect("response should build")
+    }
+}
diff --git a/src/types/v1alpha1.rs b/src/types/v1alpha1.rs
index 36d11f6..8940c44 100755
--- a/src/types/v1alpha1.rs
+++ b/src/types/v1alpha1.rs
@@ -215,6 +215,27 @@ mod tenant_provisioning_crd_tests {
             spec["properties"]["users"]["items"]["properties"]["policies"]["maxItems"],
             json!(MAX_POLICIES_PER_USER)
         );
+        assert_eq!(
+            spec["properties"]["users"]["items"]["properties"]["credsSecret"]["type"],
+            json!("object")
+        );
+        assert_eq!(
+            spec["properties"]["users"]["items"]["properties"]["credsSecret"]["properties"]["name"]
+                ["minLength"],
+            json!(1)
+        );
+        assert_eq!(
+            spec["properties"]["users"]["items"]["properties"]["credsSecret"]["required"],
+            json!(["name"])
+        );
+        assert!(
+            spec["properties"]["users"]["items"]["required"]
+                .as_array()
+                .expect("user required fields are present")
+                .iter()
+                .all(|field| field != "credsSecret"),
+            "user credsSecret must remain optional"
+        );
         assert_eq!(
             spec["properties"]["buckets"]["x-kubernetes-list-type"],
             json!("map")
@@ -247,6 +268,16 @@ mod tenant_provisioning_crd_tests {
             status["properties"]["provisioning"]["properties"]["phase"]["enum"],
             json!(["Pending", "Ready", "Failed", null])
         );
+        assert_eq!(
+            status["properties"]["provisioning"]["properties"]["users"]["items"]["properties"]["observedSecretName"]
+                ["type"],
+            json!("string")
+        );
+        assert_eq!(
+            status["properties"]["provisioning"]["properties"]["users"]["items"]["properties"]["observedSecretName"]
+                ["nullable"],
+            json!(true)
+        );
 
         assert!(
             spec["properties"]["policies"]["x-kubernetes-validations"].is_null(),
@@ -261,6 +292,20 @@ mod tenant_provisioning_crd_tests {
                 .any(|rule| rule["message"]
                     == json!("user policies must contain at least one policy"))
         );
+        assert!(
+            user_validations
+                .as_array()
+                .expect("user validations are present")
+                .iter()
+                .any(|rule| {
+                    rule["message"]
+                        == json!("user credential Secret references must be unique")
+                        && rule["rule"]
+                            == json!(
+                                "self.all(x, self.exists_one(y, (has(x.credsSecret) ? x.credsSecret.name : x.name) == (has(y.credsSecret) ? y.credsSecret.name : y.name)))"
+                            )
+                })
+        );
         let user_policy_validations = &spec["properties"]["users"]["items"]["properties"]["policies"]
             ["x-kubernetes-validations"];
         assert!(
diff --git a/src/types/v1alpha1/encryption.rs b/src/types/v1alpha1/encryption.rs
index 3f7885d..4edc0c9 100644
--- a/src/types/v1alpha1/encryption.rs
+++ b/src/types/v1alpha1/encryption.rs
@@ -16,6 +16,7 @@ use k8s_openapi::api::core::v1 as corev1;
 use k8s_openapi::schemars::JsonSchema;
 use kube::KubeSchema;
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeSet;
 
 /// KMS backend type for server-side encryption.
 ///
@@ -129,6 +130,26 @@ pub struct EncryptionConfig {
     pub default_key_id: Option<String>,
 }
 
+impl EncryptionConfig {
+    pub(crate) fn referenced_secret_names(&self) -> BTreeSet<String> {
+        let mut names = BTreeSet::new();
+        if let Some(secret) = &self.kms_secret
+            && !secret.name.is_empty()
+        {
+            names.insert(secret.name.clone());
+        }
+        if let Some(secret) = self
+            .local
+            .as_ref()
+            .and_then(|local| local.master_key_secret_ref.as_ref())
+            && !secret.name.is_empty()
+        {
+            names.insert(secret.name.clone());
+        }
+        names
+    }
+}
+
 /// Pod SecurityContext overrides for all RustFS pods in this Tenant.
 ///
 /// Overrides the default Pod SecurityContext (`runAsUser` / `runAsGroup` / `fsGroup` = 10001).
diff --git a/src/types/v1alpha1/provisioning.rs b/src/types/v1alpha1/provisioning.rs
index 9376c7e..1921a4e 100644
--- a/src/types/v1alpha1/provisioning.rs
+++ b/src/types/v1alpha1/provisioning.rs
@@ -15,12 +15,14 @@
 use kube::KubeSchema;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeSet;
 use utoipa::ToSchema;
 
 pub(crate) const MAX_CONFIG_MAP_REF_NAME_LENGTH: u32 = 253;
 pub(crate) const MAX_CONFIG_MAP_REF_KEY_LENGTH: u32 = 253;
 pub(crate) const MAX_PROVISIONING_POLICY_NAME_LENGTH: u32 = 253;
 pub(crate) const MAX_PROVISIONING_USER_NAME_LENGTH: u32 = 253;
+pub(crate) const MAX_USER_CREDENTIALS_SECRET_NAME_LENGTH: u32 = 253;
 pub(crate) const MAX_POLICIES_PER_USER: u32 = 64;
 pub(crate) const MAX_USER_POLICY_NAME_LENGTH: u32 = 253;
 pub(crate) const MIN_BUCKET_NAME_LENGTH: u32 = 3;
@@ -65,12 +67,24 @@ pub struct ProvisioningPolicy {
     pub deletion_policy: ProvisioningDeletionPolicy,
 }
 
+/// Reference to a user credentials Secret in the Tenant namespace.
+#[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct UserCredentialsSecretRef {
+    #[schemars(length(min = 1, max = MAX_USER_CREDENTIALS_SECRET_NAME_LENGTH))]
+    pub name: String,
+}
+
 #[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct ProvisioningUser {
     #[schemars(length(min = 1, max = MAX_PROVISIONING_USER_NAME_LENGTH), regex(pattern = r"^\S+$"))]
     pub name: String,
 
+    /// Optional credentials Secret reference. Defaults to a Secret named after the user.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub creds_secret: Option<UserCredentialsSecretRef>,
+
     /// Canned policies to map directly to this user.
     #[schemars(
         length(min = 1, max = MAX_POLICIES_PER_USER),
@@ -84,6 +98,29 @@ pub struct ProvisioningUser {
     pub deletion_policy: ProvisioningDeletionPolicy,
 }
 
+impl ProvisioningUser {
+    /// Resolves the credentials Secret name while preserving the legacy same-name convention.
+    pub fn credentials_secret_name(&self) -> &str {
+        self.creds_secret
+            .as_ref()
+            .map_or(self.name.as_str(), |reference| reference.name.as_str())
+    }
+}
+
+/// Returns credentials Secret names selected by more than one provisioning user.
+pub(crate) fn duplicate_user_credentials_secret_names(
+    users: &[ProvisioningUser],
+) -> BTreeSet<&str> {
+    let mut seen = BTreeSet::new();
+    users
+        .iter()
+        .filter_map(|user| {
+            let secret_name = user.credentials_secret_name();
+            (!seen.insert(secret_name)).then_some(secret_name)
+        })
+        .collect()
+}
+
 #[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, ToSchema, Default, PartialEq, Eq)]
 #[serde(rename_all = "camelCase")]
 pub struct ProvisioningBucket {
@@ -109,3 +146,66 @@ impl ProvisioningBucket {
         self.object_lock.unwrap_or(false)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        ProvisioningUser, UserCredentialsSecretRef, duplicate_user_credentials_secret_names,
+    };
+    use std::collections::BTreeSet;
+
+    #[test]
+    fn user_credentials_secret_defaults_to_user_name() {
+        let user: ProvisioningUser = serde_json::from_value(serde_json::json!({
+            "name": "app-user",
+            "policies": ["app-readwrite"]
+        }))
+        .expect("legacy ProvisioningUser should deserialize");
+
+        assert_eq!(user.credentials_secret_name(), "app-user");
+        let value = serde_json::to_value(user).expect("ProvisioningUser should serialize");
+        assert!(value.get("credsSecret").is_none());
+    }
+
+    #[test]
+    fn user_credentials_secret_uses_explicit_reference() {
+        let user = ProvisioningUser {
+            name: "app-user".to_string(),
+            creds_secret: Some(UserCredentialsSecretRef {
+                name: "rustfs-user-app-user".to_string(),
+            }),
+            ..Default::default()
+        };
+
+        assert_eq!(user.credentials_secret_name(), "rustfs-user-app-user");
+    }
+
+    #[test]
+    fn duplicate_credentials_secret_names_include_legacy_resolution() {
+        let users = [
+            ProvisioningUser {
+                name: "shared-secret".to_string(),
+                ..Default::default()
+            },
+            ProvisioningUser {
+                name: "app-user".to_string(),
+                creds_secret: Some(UserCredentialsSecretRef {
+                    name: "shared-secret".to_string(),
+                }),
+                ..Default::default()
+            },
+            ProvisioningUser {
+                name: "report-user".to_string(),
+                creds_secret: Some(UserCredentialsSecretRef {
+                    name: "report-user-secret".to_string(),
+                }),
+                ..Default::default()
+            },
+        ];
+
+        assert_eq!(
+            duplicate_user_credentials_secret_names(&users),
+            BTreeSet::from(["shared-secret"])
+        );
+    }
+}
diff --git a/src/types/v1alpha1/status/provisioning.rs b/src/types/v1alpha1/status/provisioning.rs
index 0fc2737..af976bb 100644
--- a/src/types/v1alpha1/status/provisioning.rs
+++ b/src/types/v1alpha1/status/provisioning.rs
@@ -96,6 +96,9 @@ pub struct ProvisioningItemStatus {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub observed_secret_resource_version: Option<String>,
 
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub observed_secret_name: Option<String>,
+
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub last_applied_access_key_hash: Option<String>,
 
diff --git a/src/types/v1alpha1/tenant.rs b/src/types/v1alpha1/tenant.rs
index 4789de7..75bd40c 100755
--- a/src/types/v1alpha1/tenant.rs
+++ b/src/types/v1alpha1/tenant.rs
@@ -27,6 +27,7 @@ use k8s_openapi::apimachinery::pkg::apis::meta::v1 as metav1;
 use kube::{CustomResource, KubeSchema, Resource, ResourceExt};
 use serde::{Deserialize, Serialize};
 use snafu::OptionExt;
+use std::collections::BTreeSet;
 
 // Submodules for resource factory methods
 mod helper;
@@ -38,6 +39,7 @@ pub(crate) const MAX_TENANT_POOLS: u32 = 32;
 pub(crate) const MAX_TENANT_POLICIES: u32 = 256;
 pub(crate) const MAX_TENANT_USERS: u32 = 256;
 pub(crate) const MAX_TENANT_BUCKETS: u32 = 1024;
+pub(crate) const RUSTFS_TENANT_LABEL: &str = "rustfs.tenant";
 
 /// Reference to one key in a namespaced Kubernetes Secret.
 #[derive(Deserialize, Serialize, Clone, Debug, KubeSchema, Default, PartialEq)]
@@ -205,6 +207,7 @@ pub struct TenantSpec {
         extend("x-kubernetes-list-type" = "map", "x-kubernetes-list-map-keys" = ["name"])
     )]
     #[x_kube(validation = Rule::new("self.all(x, has(x.policies) && x.policies.size() > 0)").message("user policies must contain at least one policy"))]
+    #[x_kube(validation = Rule::new("self.all(x, self.exists_one(y, (has(x.credsSecret) ? x.credsSecret.name : x.name) == (has(y.credsSecret) ? y.credsSecret.name : y.name)))").message("user credential Secret references must be unique"))]
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub users: Vec<ProvisioningUser>,
 
@@ -229,6 +232,50 @@ pub struct TenantSpec {
     pub security_context: Option<PodSecurityContextOverride>,
 }
 
+impl TenantSpec {
+    /// Returns every Kubernetes Secret name referenced by the Tenant specification.
+    pub(crate) fn referenced_secret_names(&self) -> BTreeSet<String> {
+        let mut names = BTreeSet::new();
+        {
+            let mut insert = |name: &str| {
+                if !name.is_empty() {
+                    names.insert(name.to_string());
+                }
+            };
+
+            if let Some(secret) = &self.image_pull_secret {
+                insert(&secret.name);
+            }
+            for env in &self.env {
+                if let Some(secret) = env
+                    .value_from
+                    .as_ref()
+                    .and_then(|source| source.secret_key_ref.as_ref())
+                {
+                    insert(&secret.name);
+                }
+            }
+            if let Some(secret) = &self.creds_secret {
+                insert(&secret.name);
+            }
+            if let Some(secret) = &self.rpc_secret {
+                insert(&secret.name);
+            }
+            for user in &self.users {
+                insert(user.credentials_secret_name());
+            }
+        }
+
+        if let Some(tls) = &self.tls {
+            names.extend(tls.referenced_secret_names());
+        }
+        if let Some(encryption) = &self.encryption {
+            names.extend(encryption.referenced_secret_names());
+        }
+        names
+    }
+}
+
 impl Tenant {
     pub fn namespace(&self) -> Result<String, types::error::Error> {
         ResourceExt::namespace(self).context(NoNamespaceSnafu)
@@ -288,7 +335,7 @@ impl Tenant {
                 "app.kubernetes.io/managed-by".to_owned(),
                 "rustfs-operator".to_owned(),
             ),
-            ("rustfs.tenant".to_owned(), self.name()),
+            (RUSTFS_TENANT_LABEL.to_owned(), self.name()),
         ]
         .into_iter()
         .collect()
@@ -309,7 +356,7 @@ impl Tenant {
     /// Returns selector labels for Services and StatefulSets.
     /// These should be a stable subset of the full labels.
     pub(crate) fn selector_labels(&self) -> std::collections::BTreeMap<String, String> {
-        [("rustfs.tenant".to_owned(), self.name())]
+        [(RUSTFS_TENANT_LABEL.to_owned(), self.name())]
             .into_iter()
             .collect()
     }
@@ -467,9 +514,84 @@ pub fn validate_dns1035_label(name: &str) -> Result<(), types::error::Error> {
 
 #[cfg(test)]
 mod tests {
+    use super::TenantSpec;
     use crate::types::v1alpha1::status::pool::PoolState;
     use k8s_openapi::api::apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetStatus};
     use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+    use std::collections::BTreeSet;
+
+    #[test]
+    fn referenced_secret_names_cover_all_tenant_secret_sources() {
+        let spec: TenantSpec = serde_json::from_value(serde_json::json!({
+            "pools": [],
+            "imagePullSecret": { "name": "image-pull" },
+            "env": [{
+                "name": "OIDC_CLIENT_SECRET",
+                "valueFrom": {
+                    "secretKeyRef": { "name": "env-secret", "key": "client-secret" }
+                }
+            }],
+            "credsSecret": { "name": "admin-creds" },
+            "rpcSecret": { "name": "rpc-auth", "key": "rpc-secret" },
+            "users": [
+                { "name": "legacy-user", "policies": ["readwrite"] },
+                {
+                    "name": "explicit-user",
+                    "credsSecret": { "name": "explicit-user-creds" },
+                    "policies": ["readwrite"]
+                }
+            ],
+            "tls": {
+                "certManager": {
+                    "secretName": "default-tls",
+                    "caTrust": {
+                        "caSecretRef": { "name": "default-ca", "key": "ca.crt" }
+                    }
+                },
+                "certificates": [{
+                    "name": "public",
+                    "certManager": {
+                        "secretName": "public-tls",
+                        "caTrust": {
+                            "clientCaSecretRef": {
+                                "name": "public-client-ca",
+                                "key": "ca.crt"
+                            }
+                        }
+                    }
+                }],
+                "caTrust": {
+                    "caSecretRef": { "name": "tenant-ca", "key": "ca.crt" }
+                }
+            },
+            "encryption": {
+                "kmsSecret": { "name": "vault-token" },
+                "local": {
+                    "masterKeySecretRef": { "name": "local-master-key", "key": "key" }
+                }
+            }
+        }))
+        .expect("TenantSpec should deserialize");
+
+        assert_eq!(
+            spec.referenced_secret_names(),
+            BTreeSet::from([
+                "admin-creds".to_string(),
+                "default-ca".to_string(),
+                "default-tls".to_string(),
+                "env-secret".to_string(),
+                "explicit-user-creds".to_string(),
+                "image-pull".to_string(),
+                "legacy-user".to_string(),
+                "local-master-key".to_string(),
+                "public-client-ca".to_string(),
+                "public-tls".to_string(),
+                "rpc-auth".to_string(),
+                "tenant-ca".to_string(),
+                "vault-token".to_string(),
+            ])
+        );
+    }
 
     fn statefulset_with_status(
         generation: i64,
diff --git a/src/types/v1alpha1/tls.rs b/src/types/v1alpha1/tls.rs
index 965b14f..6b021b3 100644
--- a/src/types/v1alpha1/tls.rs
+++ b/src/types/v1alpha1/tls.rs
@@ -17,7 +17,7 @@ use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString;
 use k8s_openapi::schemars::JsonSchema;
 use kube::KubeSchema;
 use serde::{Deserialize, Serialize};
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 
 pub const DEFAULT_TLS_MOUNT_PATH: &str = "/var/run/rustfs/tls";
 pub const TLS_HASH_ANNOTATION: &str = "operator.rustfs.com/tls-hash";
@@ -255,6 +255,45 @@ impl TlsConfig {
             .and_then(|certificate| certificate.cert_manager.ca_trust.clone())
             .unwrap_or_default()
     }
+
+    pub(crate) fn referenced_secret_names(&self) -> BTreeSet<String> {
+        let mut names = BTreeSet::new();
+        if let Some(cert_manager) = &self.cert_manager {
+            extend_cert_manager_secret_names(&mut names, cert_manager);
+        }
+        for certificate in &self.certificates {
+            extend_cert_manager_secret_names(&mut names, &certificate.cert_manager);
+        }
+        if let Some(ca_trust) = &self.ca_trust {
+            extend_ca_trust_secret_names(&mut names, ca_trust);
+        }
+        names
+    }
+}
+
+fn extend_cert_manager_secret_names(names: &mut BTreeSet<String>, config: &CertManagerTlsConfig) {
+    if let Some(name) = &config.secret_name
+        && !name.is_empty()
+    {
+        names.insert(name.clone());
+    }
+    if let Some(ca_trust) = &config.ca_trust {
+        extend_ca_trust_secret_names(names, ca_trust);
+    }
+}
+
+fn extend_ca_trust_secret_names(names: &mut BTreeSet<String>, config: &CaTrustConfig) {
+    for secret in [
+        config.ca_secret_ref.as_ref(),
+        config.client_ca_secret_ref.as_ref(),
+    ]
+    .into_iter()
+    .flatten()
+    {
+        if !secret.name.is_empty() {
+            names.insert(secret.name.clone());
+        }
+    }
 }
 
 #[cfg(test)]