Repository: risingwavelabs/risingwave
File: src/frontend/src/handler/alter_streaming_rate_limit.rs (Rust)

Code before:
                    "\"{table_name}\" is not a CDC table",
                ))
                .into());
            }
            (StatementType::ALTER_TABLE, table.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Sink) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            if sink.target_table.is_some() {
                bail!("ALTER SINK_RATE_LIMIT is not for sink into table")
            }
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Backfill) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        _ => bail!(
            "Unsupported throttle target: {:?} and throttle type: {:?}",
            throttle_target,
            throttle_type
        ),
    };
    execute_with_long_running_notification(
        handle_alter_streaming_rate_limit_by_id(
            &session,
            throttle_target,
            throttle_type,
            id,
            rate_limit,
            stmt_type,
        ),
        &session,
        "ALTER STREAMING RATE LIMIT",
        LongRunningNotificationAction::SuggestRecover,
    )
    .await
}

pub async fn handle_alter_streaming_rate_limit_by_id(
    session: &SessionImpl,
    throttle_target: PbThrottleTarget,
    throttle_type: PbThrottleType,
    id: u32,
    rate_limit: i32,

Review comment (security):
Missing privilege check for sink backfill rate limit. Unlike the sink rate limit case at lines 110-119 where there's a `check_privilege_for_drop_alter` call, this branch gets the sink but doesn't verify the user has privileges to alter it. The check should be added after line 123 before line 124.

Code after:
                check_table_mismatch(&reader, &session, db_name, schema_path, &real_table_name)?;
            if table.cdc_table_type.is_none() {
                return Err(ErrorCode::InvalidInputSyntax(format!(
                    "\"{table_name}\" is not a CDC table",
                ))
                .into());
            }
            (StatementType::ALTER_TABLE, table.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Sink) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            if sink.target_table.is_some() {
                bail!("ALTER SINK_RATE_LIMIT is not for sink into table")
            }
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        (PbThrottleTarget::Sink, PbThrottleType::Backfill) => {
            let reader = session.env().catalog_reader().read_guard();
            let (sink, schema_name) =
                reader.get_any_sink_by_name(db_name, schema_path, &real_table_name)?;
            session.check_privilege_for_drop_alter(schema_name, &**sink)?;
            (StatementType::ALTER_SINK, sink.id.as_raw_id())
        }
        _ => bail!(
            "Unsupported throttle target: {:?} and throttle type: {:?}",
            throttle_target,
            throttle_type
        ),
    };
    execute_with_long_running_notification(
        handle_alter_streaming_rate_limit_by_id(
            &session,
            throttle_target,
            throttle_type,
            id,
            rate_limit,
            stmt_type,
        ),
        &session,
        "ALTER STREAMING RATE LIMIT",
        LongRunningNotificationAction::SuggestRecover,
    )
    .await
}

pub async fn handle_alter_streaming_rate_limit_by_id(
    session: &SessionImpl,
    throttle_target: PbThrottleTarget,