Compare commits

..

1 Commits

Author SHA1 Message Date
唐小鸭 d5ba6b4e16 fix(admin): advertise IAM admin capabilities in runtime capabilities
The v4 runtime capabilities response carried no admin.iam.* entries, so
the rc client's capability gate rejected policy detach even though the
detach route is implemented (rustfs/backlog#1900). Advertise the IAM
admin capability set as a flat named list whose statuses are derived
from the public admin route inventory, so the advertisement tracks the
actually registered routes instead of a hardcoded claim.
2026-08-21 18:42:09 +08:00
14 changed files with 106 additions and 25 deletions
@@ -866,7 +866,7 @@ impl BucketTargetSys {
return Some(cli);
}
// TODO(backlog): spawn an async task to proactively reload the replication target
// TODO: spawn a task to reload the target
if self.is_reloading_target(bucket, arn).await {
return None;
}
@@ -454,7 +454,7 @@ impl S3PeerSys {
}
}
topology_complete &= bucket_map.values().all(|count| *count >= quorum);
// TODO(backlog): integrate MRF backlog stats into scanner bucket listing
// TODO: MRF
}
let mut buckets: Vec<BucketInfo> = result_map.into_values().collect();
@@ -2406,7 +2406,7 @@ impl DiskAPI for RemoteDisk {
return errors;
}
// TODO(backlog): replace string errors with typed `StorageError` variants
// TODO: use Error not string
let result = self
.execute_with_timeout(
+1 -1
View File
@@ -249,7 +249,7 @@ impl Sets {
self.connect_disks().await;
// TODO(backlog): make monitor_and_connect interval configurable instead of hardcoded 15s
// TODO: config interval
let mut interval = tokio::time::interval(Duration::from_secs(15));
loop {
tokio::select! {
+7 -7
View File
@@ -5215,8 +5215,8 @@ impl LocalDisk {
let cache = Cache::new(update_fn, Duration::from_secs(1), Opts::default());
// TODO(backlog): add O_DIRECT I/O support for performance-critical paths
// TODO(backlog): populate DiskInfo in constructor
// TODO: DIRECT support
// TODD: DiskInfo
let mut disk = Self {
root: root.clone(),
publication_root,
@@ -5751,7 +5751,7 @@ impl LocalDisk {
// return Ok(());
// TODO(backlog): make disk space checks and trash cleanup event-driven instead of poll-based
// TODO: async notifications for disk space checks and trash cleanup
let trash_path = self.io_get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
// if let Some(parent) = trash_path.parent() {
@@ -5997,7 +5997,7 @@ impl LocalDisk {
#[hotpath::measure(impl_type = "LocalDisk")]
async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
// TODO(backlog): add configurable timeout for read_all_data operations
// TODO: timeout support
let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;
Ok(data)
}
@@ -6674,7 +6674,7 @@ impl LocalDisk {
return Ok(());
}
// TODO(backlog): add directory listing lock to prevent concurrent enumeration
// TODO: add lock
let stall = opts.stall_timeout_duration();
@@ -8796,7 +8796,7 @@ impl DiskAPI for LocalDisk {
Ok(entries)
}
// TODO(backlog): support io.writer cancellation and early termination in walk_dir
// FIXME: TODO: io.writer TODO cancel
#[tracing::instrument(level = "trace", skip_all)]
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
self.wait_for_startup_cleanup().await;
@@ -9880,7 +9880,7 @@ impl DiskAPI for LocalDisk {
);
return Err(e);
}
// TODO(backlog): add post-setup disk health verification
// TODO: health check
}
Ok(())
}
+2 -2
View File
@@ -249,7 +249,7 @@ impl PoolEndpointList {
endpoint.set_set_index(0);
endpoint.set_disk_index(0);
// TODO(backlog): check for cross-device mounts in single-drive setup
// TODO Check for cross device mounts if any.
return Ok(Self {
inner: vec![Endpoints::from(vec![endpoint])],
@@ -264,7 +264,7 @@ impl PoolEndpointList {
// Convert args to endpoints
let mut eps = Endpoints::try_from(set_layout.as_slice())?;
// TODO(backlog): check for cross-device mounts in multi-pool setup
// TODO Check for cross device mounts if any.
for (disk_idx, ep) in eps.as_mut().iter_mut().enumerate() {
ep.set_pool_index(pool_idx);
+1 -1
View File
@@ -1091,7 +1091,7 @@ impl ObjectInfo {
}
};
// TODO(backlog): handle VersionPurgeStatus in object listing
// TODO:VersionPurgeStatus
let versioned = vcfg.clone().map(|v| v.0.versioned(&entry.name)).unwrap_or_default();
objects.push(ObjectInfo::from_file_info(&fi, bucket, &entry.name, versioned));
+2 -2
View File
@@ -1575,7 +1575,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let parts_metadata = vec![fi.clone(); disks.len()];
if !user_defined.contains_key("content-type") {
// TODO(backlog): detect content-type from part data when header is missing
// TODO: get content-type
}
if let Some(sc) = user_defined.get(AMZ_STORAGE_CLASS)
@@ -1971,7 +1971,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
}
// TODO(backlog): integrate encryption verification during complete multipart
// TODO: crypto
if (i < uploaded_parts.len() - 1)
&& !(opts.data_movement && ext_part.actual_size < 0)
+3 -3
View File
@@ -6161,7 +6161,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
join_all(rollback_futures).await;
// TODO(backlog): support partial object deletion for multi-part objects
// TODO: add_partial
if let Some(api) = opts.tier_delete_journal_api.as_ref() {
for (idx, je) in persisted_journal_entries {
@@ -6371,7 +6371,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
}
}
// TODO(backlog): integrate lifecycle evaluation before object deletion
// TODO: Lifecycle
let mut version_found = true;
// delete_object_version below derives its own majority quorum from the
@@ -6465,7 +6465,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
mark_deleted: mark_delete,
mod_time: Some(mod_time),
replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta),
..Default::default() // TODO(backlog): populate transition state on delete markers
..Default::default() // TODO: Transition
};
fi.set_tier_free_version_id(&find_vid.to_string());
+1 -1
View File
@@ -601,7 +601,7 @@ impl ECStore {
#[instrument(skip(self))]
pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
// TODO(backlog): support cached bucket listing via opts.cached
// TODO: opts.cached
let mut buckets = self.peer_sys.list_bucket(opts).await?;
+2 -2
View File
@@ -4673,7 +4673,7 @@ async fn gather_results(
entry.name = entry.name.replace("\\", "/");
}
// TODO(backlog): integrate rx.recv() for incremental listing results
// TODO: rx.recv()
if let Some(marker) = &opts.marker
&& ((!opts.include_marker && &entry.name <= marker) || (opts.include_marker && &entry.name < marker))
@@ -4703,7 +4703,7 @@ async fn gather_results(
continue;
}
// TODO(backlog): integrate lifecycle evaluation during object listing
// TODO: Lifecycle
entries.push(Some(entry));
candidate_entries += 1;
+2 -2
View File
@@ -332,7 +332,7 @@ impl ECStore {
let expected_incarnation_id = opts.expected_bucket_incarnation_id;
if request.prefix.is_empty() {
// TODO(backlog): return cached multipart listing when prefix is empty
// TODO: return from cache
}
if self.single_pool() {
@@ -610,7 +610,7 @@ impl ECStore {
let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
let opts = &opts;
// TODO(backlog): defer DeleteUploadID to background for faster abort response
// TODO: defer DeleteUploadID
if self.single_pool() {
return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await;
+1 -1
View File
@@ -385,7 +385,7 @@ impl ECStore {
}
pub(super) async fn is_suspended(&self, idx: usize) -> bool {
// TODO(backlog): acquire pool metadata lock for consistent suspension check
// TODO: LOCK
let pool_meta = self.pool_meta.read().await;
+81
View File
@@ -70,6 +70,12 @@ const SITE_REPLICATION_EDIT_ROUTE: &str = "/rustfs/admin/v3/site-replication/edi
const SITE_REPLICATION_RESYNC_ROUTE: &str = "/rustfs/admin/v3/site-replication/resync/op";
const SITE_REPLICATION_REPAIR_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair";
const SITE_REPLICATION_REPAIR_STATUS_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair/status";
const IAM_POLICY_ATTACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/attach";
const IAM_POLICY_DETACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/detach";
const IAM_POLICY_ENTITIES_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy-entities";
const IAM_ACCESS_KEYS_BULK_ROUTE: &str = "/rustfs/admin/v3/list-access-keys-bulk";
const IAM_ACCESS_KEYS_BULK_LDAP_ROUTE: &str = "/rustfs/admin/v3/idp/ldap/list-access-keys-bulk";
const IAM_ACCESS_KEYS_BULK_OPENID_ROUTE: &str = "/rustfs/admin/v3/idp/openid/list-access-keys-bulk";
macro_rules! log_system_request_rejected {
($operation:expr, $reason:expr) => {
@@ -661,9 +667,24 @@ pub struct RuntimeCapabilitiesSummary {
pub manual_transition_jobs: CapabilityStatus,
}
/// One named admin capability advertised to management clients
/// (rustfs/backlog#1900). `name` is a cross-repo wire contract: the rc
/// client gates commands on these exact strings (see rustfs/cli
/// `IAM_POLICY_DETACH_CAPABILITY` etc.), so entries may be added but
/// existing names must never be renamed or removed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct AdvertisedAdminCapability {
pub name: &'static str,
pub status: CapabilityStatus,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct RuntimeCapabilitiesResponse {
pub summary: RuntimeCapabilitiesSummary,
/// Additive field: absent in responses from older servers, so clients
/// must treat a missing list as "no dynamic advertisement" and fall
/// back to their pinned per-version contract.
pub advertised: Vec<AdvertisedAdminCapability>,
pub replication: ReplicationCapabilities,
pub manual_transition_jobs: ManualTransitionJobCapabilities,
pub diagnostic_probes: DiagnosticProbeCapabilities,
@@ -986,6 +1007,7 @@ pub(crate) async fn build_runtime_capabilities_response()
Ok(RuntimeCapabilitiesResponse {
summary,
advertised: advertised_admin_capabilities(),
replication: ReplicationCapabilities::current(),
manual_transition_jobs: ManualTransitionJobCapabilities::current(),
diagnostic_probes: DiagnosticProbeCapabilities::current(),
@@ -1077,6 +1099,23 @@ fn admin_route_capability(method: HttpMethod, path: &str) -> CapabilityStatus {
admin_route_capability_from_inventory(method, path, ADMIN_ROUTE_POLICY_SPECS, DEFERRED_ADMIN_ROUTE_POLICIES)
}
fn advertised_admin_capabilities() -> Vec<AdvertisedAdminCapability> {
[
("admin.iam.policy-attach", HttpMethod::Post, IAM_POLICY_ATTACH_ROUTE),
("admin.iam.policy-detach", HttpMethod::Post, IAM_POLICY_DETACH_ROUTE),
("admin.iam.policy-entities", HttpMethod::Get, IAM_POLICY_ENTITIES_ROUTE),
("admin.iam.access-keys-bulk", HttpMethod::Get, IAM_ACCESS_KEYS_BULK_ROUTE),
("admin.iam.access-keys-bulk.ldap", HttpMethod::Get, IAM_ACCESS_KEYS_BULK_LDAP_ROUTE),
("admin.iam.access-keys-bulk.openid", HttpMethod::Get, IAM_ACCESS_KEYS_BULK_OPENID_ROUTE),
]
.into_iter()
.map(|(name, method, route)| AdvertisedAdminCapability {
name,
status: admin_route_capability(method, route),
})
.collect()
}
fn admin_route_capability_from_inventory(
method: HttpMethod,
path: &str,
@@ -1239,6 +1278,48 @@ mod tests {
);
}
/// Wire-contract pin (rustfs/backlog#1900): the rc client keys its
/// command gates on these exact capability names, and parses each
/// entry as `{name, status: {state, reason?}}`. Renaming or dropping
/// a name silently disables the corresponding rc command.
#[tokio::test]
async fn runtime_capabilities_response_advertises_iam_capabilities() {
let response = build_runtime_capabilities_response()
.await
.expect("runtime capabilities response should build");
let expected_supported = [
"admin.iam.policy-attach",
"admin.iam.policy-detach",
"admin.iam.policy-entities",
"admin.iam.access-keys-bulk",
"admin.iam.access-keys-bulk.ldap",
"admin.iam.access-keys-bulk.openid",
];
for name in expected_supported {
let entry = response
.advertised
.iter()
.find(|capability| capability.name == name)
.unwrap_or_else(|| panic!("{name} must be advertised"));
assert_eq!(entry.status.state, CapabilityState::Supported, "{name} must be supported");
}
let mut names: Vec<&str> = response.advertised.iter().map(|capability| capability.name).collect();
let total = names.len();
names.sort_unstable();
names.dedup();
assert_eq!(names.len(), total, "advertised capability names must be unique");
let serialized = serde_json::to_value(&response).expect("response should serialize");
let advertised = serialized["advertised"].as_array().expect("advertised must be an array");
let detach = advertised
.iter()
.find(|entry| entry["name"] == "admin.iam.policy-detach")
.expect("serialized detach entry must exist");
assert_eq!(detach["status"]["state"], "supported");
}
#[tokio::test]
async fn runtime_capabilities_response_reports_missing_topology_before_storage_init() {
let response = build_runtime_capabilities_response()