[pbs-devel] [PATCH proxmox{, -backup} v3 0/6] add user specific rate-limits

[pbs-devel] [PATCH proxmox{, -backup} v3 0/6] add user specific rate-limits

[Hannes Laimer]

When a connection is accepted we create a shared tag handle for its
rate-limited stream. The REST layer clears that handle before every
request. Once a request authenticates successfully, we push a
User(...) tag with the auth ID. Failed or unauthenticated requests
leave the tag list empty. RateLimitedStream watches that handle and
forces an immediate limiter refresh whenever the tag set changes so
user-specific throttles take effect right away.

Currently rules with a user specified take priority over others. So:
user > IP only > neither, in case two rules match.

If users and networks are specified, the rule only applies if both
match. So, Any of the specified user connect from any of the specified
network.

And all of this ofc still only if the given timeframe matches.

I did also test this with a basic nginx reverse proxy configured with
`keepalive 32`, I didn't run into problems using this setup.

v3, thanks @Chris!:
 - simplify code by passing the taglist to the callback, as sugested by
   Chris
 - mention potential future use-case in commit message
 - created documented type for 3-tuple and inlined var for printing

v2, thanks @Chris!:
 - fix problem with tag staying on connection after request finishes,
   and with when it would be set in first place
 - use a more generic tag-list on the connection, this is more general
 - tag is now an enum, like chris suggested, this should make it
   somewhat easy to extend if we at some point should want to


proxmox:

Hannes Laimer (3):
  pbs-api-types: allow traffic-control rules to match users
  http: track user tag updates on rate-limited streams
  rest-server: propagate rate-limit tags from authenticated users

 pbs-api-types/src/traffic_control.rs    |   9 ++
 proxmox-http/src/lib.rs                 |   2 +-
 proxmox-http/src/rate_limited_stream.rs |  40 ++++++-
 proxmox-rest-server/src/connection.rs   |  11 +-
 proxmox-rest-server/src/rest.rs         | 137 +++++++++++++++++++++++-
 5 files changed, 186 insertions(+), 13 deletions(-)


proxmox-backup:

Hannes Laimer (3):
  api: taffic-control: update/delete users on rule correctly
  traffic-control: handle users specified in a rule correctly
  ui: traffic-control: add users field in edit form and list

 src/api2/config/traffic_control.rs |   8 +++
 src/bin/proxmox-backup-proxy.rs    |  12 +++-
 src/traffic_control_cache.rs       | 105 +++++++++++++++++++++++++----
 www/config/TrafficControlView.js   |   7 ++
 www/window/TrafficControlEdit.js   |  18 +++++
 5 files changed, 136 insertions(+), 14 deletions(-)


Summary over all repositories:
  10 files changed, 322 insertions(+), 27 deletions(-)

-- 
Generated by git-murpp 0.8.1


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox v3 1/3] pbs-api-types: allow traffic-control rules to match users

[Hannes Laimer]

Extend traffic-control rules with an optional list of user IDs so
API traffic can be limited per user in addition to IP-based rules.

Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 pbs-api-types/src/traffic_control.rs | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/pbs-api-types/src/traffic_control.rs b/pbs-api-types/src/traffic_control.rs
index 2a359eda..12fc8c93 100644
--- a/pbs-api-types/src/traffic_control.rs
+++ b/pbs-api-types/src/traffic_control.rs
@@ -6,6 +6,7 @@ use proxmox_schema::{api, ApiType, Schema, StringSchema, Updater};
 use proxmox_schema::api_types::CIDR_SCHEMA;
 
 use crate::{DAILY_DURATION_FORMAT, PROXMOX_SAFE_ID_FORMAT, SINGLE_LINE_COMMENT_SCHEMA};
+use crate::Userid;
 
 pub const TRAFFIC_CONTROL_TIMEFRAME_SCHEMA: Schema =
     StringSchema::new("Timeframe to specify when the rule is active.")
@@ -125,6 +126,11 @@ pub struct ClientRateLimitConfig {
             },
             optional: true,
         },
+        users: {
+            type: Array,
+            items: { type: Userid },
+            optional: true,
+        },
     },
 )]
 #[derive(Clone, Serialize, Deserialize, PartialEq, Updater)]
@@ -146,6 +152,9 @@ pub struct TrafficControlRule {
     /// Enable the rule at specific times
     #[serde(skip_serializing_if = "Option::is_none")]
     pub timeframe: Option<Vec<String>>,
+    /// Rule applies to authenticated API requests of any of these users (overrides IP-only rules)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub users: Option<Vec<Userid>>,
 }
 
 #[api(
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox v3 2/3] http: track user tag updates on rate-limited streams

[Hannes Laimer]

Introduce rate-limit tags with a user variant and let rate-limited
streams hold a shared handle so callbacks can refresh limits
whenever the tag set changes.

If we decide to implement something like [1] in the future this could
potentially include group rate-limits for example.

[1] https://bugzilla.proxmox.com/show_bug.cgi?id=5867

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 proxmox-http/src/lib.rs                 |  2 +-
 proxmox-http/src/rate_limited_stream.rs | 40 +++++++++++++++++++++----
 2 files changed, 36 insertions(+), 6 deletions(-)

diff --git a/proxmox-http/src/lib.rs b/proxmox-http/src/lib.rs
index 8b6953b0..990f8f36 100644
--- a/proxmox-http/src/lib.rs
+++ b/proxmox-http/src/lib.rs
@@ -34,7 +34,7 @@ pub use rate_limiter::{RateLimit, RateLimiter, RateLimiterVec, ShareableRateLimi
 #[cfg(feature = "rate-limited-stream")]
 mod rate_limited_stream;
 #[cfg(feature = "rate-limited-stream")]
-pub use rate_limited_stream::RateLimitedStream;
+pub use rate_limited_stream::{RateLimitedStream, RateLimiterTag, RateLimiterTags};
 
 #[cfg(feature = "body")]
 mod body;
diff --git a/proxmox-http/src/rate_limited_stream.rs b/proxmox-http/src/rate_limited_stream.rs
index e24df7af..6b525591 100644
--- a/proxmox-http/src/rate_limited_stream.rs
+++ b/proxmox-http/src/rate_limited_stream.rs
@@ -15,8 +15,15 @@ use super::{RateLimiter, ShareableRateLimit};
 
 type SharedRateLimit = Arc<dyn ShareableRateLimit>;
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum RateLimiterTag {
+    User(String),
+}
+
+pub type RateLimiterTags = Vec<RateLimiterTag>;
+
 pub type RateLimiterCallback =
-    dyn Fn() -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send;
+    dyn Fn(&[RateLimiterTag]) -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send;
 
 /// A rate limited stream using [RateLimiter]
 pub struct RateLimitedStream<S> {
@@ -26,6 +33,8 @@ pub struct RateLimitedStream<S> {
     write_delay: Option<Pin<Box<Sleep>>>,
     update_limiter_cb: Option<Box<RateLimiterCallback>>,
     last_limiter_update: Instant,
+    tag_handle: Option<Arc<Mutex<RateLimiterTags>>>,
+    last_tags: Option<RateLimiterTags>,
     stream: S,
 }
 
@@ -53,6 +62,8 @@ impl<S> RateLimitedStream<S> {
             write_delay: None,
             update_limiter_cb: None,
             last_limiter_update: Instant::now(),
+            tag_handle: None,
+            last_tags: None,
             stream,
         }
     }
@@ -64,12 +75,13 @@ impl<S> RateLimitedStream<S> {
     /// Note: This function is called within an async context, so it
     /// should be fast and must not block.
     pub fn with_limiter_update_cb<
-        F: Fn() -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send + 'static,
+        F: Fn(&[RateLimiterTag]) -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send + 'static,
     >(
         stream: S,
         update_limiter_cb: F,
     ) -> Self {
-        let (read_limiter, write_limiter) = update_limiter_cb();
+        let tag_handle = Some(Arc::new(Mutex::new(Vec::new())));
+        let (read_limiter, write_limiter) = update_limiter_cb(&[]);
         Self {
             read_limiter,
             read_delay: None,
@@ -77,15 +89,29 @@ impl<S> RateLimitedStream<S> {
             write_delay: None,
             update_limiter_cb: Some(Box::new(update_limiter_cb)),
             last_limiter_update: Instant::now(),
+            tag_handle,
+            last_tags: None,
             stream,
         }
     }
 
     fn update_limiters(&mut self) {
         if let Some(ref update_limiter_cb) = self.update_limiter_cb {
-            if self.last_limiter_update.elapsed().as_secs() >= 5 {
+            let mut force_update = false;
+            let current_tags = self
+                .tag_handle
+                .as_ref()
+                .map(|handle| handle.lock().unwrap().clone());
+
+            if self.last_tags != current_tags {
+                self.last_tags = current_tags.clone();
+                force_update = true;
+            }
+
+            if force_update || self.last_limiter_update.elapsed().as_secs() >= 5 {
                 self.last_limiter_update = Instant::now();
-                let (read_limiter, write_limiter) = update_limiter_cb();
+                let tags = self.last_tags.as_ref().map(|tags| tags.as_slice()).unwrap_or(&[]);
+                let (read_limiter, write_limiter) = update_limiter_cb(tags);
                 self.read_limiter = read_limiter;
                 self.write_limiter = write_limiter;
             }
@@ -99,6 +125,10 @@ impl<S> RateLimitedStream<S> {
     pub fn inner_mut(&mut self) -> &mut S {
         &mut self.stream
     }
+
+    pub fn tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
+        self.tag_handle.as_ref().map(Arc::clone)
+    }
 }
 
 fn register_traffic(limiter: &dyn ShareableRateLimit, count: usize) -> Option<Pin<Box<Sleep>>> {
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox v3 3/3] rest-server: propagate rate-limit tags from authenticated users

[Hannes Laimer]

Tie REST connections rate-limiter callbacks to a shared tag handle
so we can push authenticated user IDs into the limiter, keeping
tags in sync whenever we clear or update them.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 proxmox-rest-server/src/connection.rs |  11 ++-
 proxmox-rest-server/src/rest.rs       | 137 +++++++++++++++++++++++++-
 2 files changed, 141 insertions(+), 7 deletions(-)

diff --git a/proxmox-rest-server/src/connection.rs b/proxmox-rest-server/src/connection.rs
index 9511b7cb..ff2ee139 100644
--- a/proxmox-rest-server/src/connection.rs
+++ b/proxmox-rest-server/src/connection.rs
@@ -24,7 +24,7 @@ use tokio_openssl::SslStream;
 use tokio_stream::wrappers::ReceiverStream;
 
 #[cfg(feature = "rate-limited-stream")]
-use proxmox_http::{RateLimitedStream, ShareableRateLimit};
+use proxmox_http::{RateLimitedStream, RateLimiterTag, ShareableRateLimit};
 
 #[cfg(feature = "rate-limited-stream")]
 pub type SharedRateLimit = Arc<dyn ShareableRateLimit>;
@@ -165,7 +165,10 @@ type InsecureClientStreamResult = Pin<Box<InsecureClientStream>>;
 type ClientStreamResult = Pin<Box<SslStream<InsecureClientStream>>>;
 
 #[cfg(feature = "rate-limited-stream")]
-type LookupRateLimiter = dyn Fn(std::net::SocketAddr) -> (Option<SharedRateLimit>, Option<SharedRateLimit>)
+type LookupRateLimiter = dyn Fn(
+        std::net::SocketAddr,
+        &[RateLimiterTag],
+    ) -> (Option<SharedRateLimit>, Option<SharedRateLimit>)
     + Send
     + Sync
     + 'static;
@@ -369,7 +372,9 @@ impl AcceptBuilder {
 
         #[cfg(feature = "rate-limited-stream")]
         let socket = match self.lookup_rate_limiter.clone() {
-            Some(lookup) => RateLimitedStream::with_limiter_update_cb(socket, move || lookup(peer)),
+            Some(lookup) => RateLimitedStream::with_limiter_update_cb(socket, move |tags| {
+                lookup(peer, tags)
+            }),
             None => RateLimitedStream::with_limiter(socket, None, None),
         };
 
diff --git a/proxmox-rest-server/src/rest.rs b/proxmox-rest-server/src/rest.rs
index b76c4bc9..6e9692ae 100644
--- a/proxmox-rest-server/src/rest.rs
+++ b/proxmox-rest-server/src/rest.rs
@@ -29,6 +29,10 @@ use tower_service::Service;
 use url::form_urlencoded;
 
 use proxmox_http::Body;
+#[cfg(feature = "rate-limited-stream")]
+use proxmox_http::{RateLimiterTag, RateLimiterTags};
+#[cfg(not(feature = "rate-limited-stream"))]
+type RateLimiterTags = ();
 use proxmox_router::{
     check_api_permission, ApiHandler, ApiMethod, HttpError, Permission, RpcEnvironment,
     RpcEnvironmentType, UserInformation,
@@ -86,10 +90,26 @@ impl RestServer {
         }
     }
 
-    pub fn api_service(&self, peer: &dyn PeerAddress) -> Result<ApiService, Error> {
+    #[cfg(not(feature = "rate-limited-stream"))]
+    pub fn api_service<T>(&self, peer: &T) -> Result<ApiService, Error>
+    where
+        T: PeerAddress + ?Sized,
+    {
+        Ok(ApiService {
+            peer: peer.peer_addr()?,
+            api_config: Arc::clone(&self.api_config),
+        })
+    }
+
+    #[cfg(feature = "rate-limited-stream")]
+    pub fn api_service<T>(&self, peer: &T) -> Result<ApiService, Error>
+    where
+        T: PeerAddress + PeerRateLimitTags + ?Sized,
+    {
         Ok(ApiService {
             peer: peer.peer_addr()?,
             api_config: Arc::clone(&self.api_config),
+            rate_limit_tags: peer.rate_limiter_tag_handle(),
         })
     }
 }
@@ -185,6 +205,11 @@ pub trait PeerAddress {
     fn peer_addr(&self) -> Result<std::net::SocketAddr, Error>;
 }
 
+#[cfg(feature = "rate-limited-stream")]
+pub trait PeerRateLimitTags {
+    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>>;
+}
+
 // tokio_openssl's SslStream requires the stream to be pinned in order to accept it, and we need to
 // accept before the peer address is requested, so let's just generally implement this for
 // Pin<Box<T>>
@@ -221,6 +246,41 @@ impl<T: PeerAddress> PeerAddress for proxmox_http::RateLimitedStream<T> {
     }
 }
 
+#[cfg(feature = "rate-limited-stream")]
+impl<T: PeerRateLimitTags> PeerRateLimitTags for Pin<Box<T>> {
+    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
+        T::rate_limiter_tag_handle(&**self)
+    }
+}
+
+#[cfg(feature = "rate-limited-stream")]
+impl<T: PeerRateLimitTags> PeerRateLimitTags for tokio_openssl::SslStream<T> {
+    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
+        self.get_ref().rate_limiter_tag_handle()
+    }
+}
+
+#[cfg(feature = "rate-limited-stream")]
+impl PeerRateLimitTags for tokio::net::TcpStream {
+    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
+        None
+    }
+}
+
+#[cfg(feature = "rate-limited-stream")]
+impl PeerRateLimitTags for tokio::net::UnixStream {
+    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
+        None
+    }
+}
+
+#[cfg(feature = "rate-limited-stream")]
+impl<T> PeerRateLimitTags for proxmox_http::RateLimitedStream<T> {
+    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
+        self.tag_handle()
+    }
+}
+
 // Helper [Service] containing the peer Address
 //
 // The lower level connection [Service] implementation on
@@ -233,6 +293,8 @@ impl<T: PeerAddress> PeerAddress for proxmox_http::RateLimitedStream<T> {
 pub struct ApiService {
     pub peer: std::net::SocketAddr,
     pub api_config: Arc<ApiConfig>,
+    #[cfg(feature = "rate-limited-stream")]
+    pub rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>>,
 }
 
 impl ApiService {
@@ -354,6 +416,10 @@ impl Service<Request<Incoming>> for ApiService {
             Some(proxied_peer) => proxied_peer,
             None => self.peer,
         };
+        #[cfg(feature = "rate-limited-stream")]
+        let rate_limit_tags = self.rate_limit_tags.clone();
+        #[cfg(not(feature = "rate-limited-stream"))]
+        let rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>> = None;
 
         let header = self.api_config
             .auth_cookie_name
@@ -368,7 +434,15 @@ impl Service<Request<Incoming>> for ApiService {
              });
 
         async move {
-            let mut response = match Arc::clone(&config).handle_request(req, &peer).await {
+            #[cfg(feature = "rate-limited-stream")]
+            if let Some(handle) = rate_limit_tags.as_ref() {
+                handle.lock().unwrap().clear();
+            }
+
+            let mut response = match Arc::clone(&config)
+                .handle_request(req, &peer, rate_limit_tags.clone())
+                .await
+            {
                 Ok(response) => response,
                 Err(err) => {
                     let (err, code) = match err.downcast_ref::<HttpError>() {
@@ -860,6 +934,8 @@ impl ApiConfig {
         self: Arc<ApiConfig>,
         req: Request<Incoming>,
         peer: &std::net::SocketAddr,
+        #[cfg_attr(not(feature = "rate-limited-stream"), allow(unused_variables))]
+        rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>>,
     ) -> Result<Response<Body>, Error> {
         let (parts, body) = req.into_parts();
         let method = parts.method.clone();
@@ -890,6 +966,8 @@ impl ApiConfig {
                     full_path: &path,
                     relative_path_components,
                     rpcenv,
+                    #[cfg(feature = "rate-limited-stream")]
+                    rate_limit_tags: rate_limit_tags.clone(),
                 })
                 .await;
         }
@@ -901,13 +979,29 @@ impl ApiConfig {
         if components.is_empty() {
             match self.check_auth(&parts.headers, &method).await {
                 Ok((auth_id, _user_info)) => {
-                    rpcenv.set_auth_id(Some(auth_id));
+                    rpcenv.set_auth_id(Some(auth_id.clone()));
+                    #[cfg(feature = "rate-limited-stream")]
+                    if let Some(handle) = rate_limit_tags.as_ref() {
+                        let mut guard = handle.lock().unwrap();
+                        guard.clear();
+                        guard.push(RateLimiterTag::User(auth_id));
+                    }
                     return Ok(self.get_index(rpcenv, parts).await);
                 }
                 Err(AuthError::Generic(_)) => {
+                    #[cfg(feature = "rate-limited-stream")]
+                    if let Some(handle) = rate_limit_tags.as_ref() {
+                        handle.lock().unwrap().clear();
+                    }
                     tokio::time::sleep_until(Instant::from_std(delay_unauth_time())).await;
                 }
-                Err(AuthError::NoData) => {}
+                Err(AuthError::NoData) =>
+                {
+                    #[cfg(feature = "rate-limited-stream")]
+                    if let Some(handle) = rate_limit_tags.as_ref() {
+                        handle.lock().unwrap().clear();
+                    }
+                }
             }
             Ok(self.get_index(rpcenv, parts).await)
         } else {
@@ -975,6 +1069,8 @@ pub struct ApiRequestData<'a> {
     full_path: &'a str,
     relative_path_components: &'a [&'a str],
     rpcenv: RestEnvironment,
+    #[cfg(feature = "rate-limited-stream")]
+    rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>>,
 }
 
 pub(crate) struct Formatted {
@@ -992,6 +1088,8 @@ impl Formatted {
             full_path,
             relative_path_components,
             mut rpcenv,
+            #[cfg(feature = "rate-limited-stream")]
+            rate_limit_tags,
         }: ApiRequestData<'_>,
     ) -> Result<Response<Body>, Error> {
         if relative_path_components.is_empty() {
@@ -1026,10 +1124,20 @@ impl Formatted {
         if auth_required {
             match config.check_auth(&parts.headers, &parts.method).await {
                 Ok((authid, info)) => {
+                    #[cfg(feature = "rate-limited-stream")]
+                    if let Some(handle) = rate_limit_tags.as_ref() {
+                        let mut guard = handle.lock().unwrap();
+                        guard.clear();
+                        guard.push(RateLimiterTag::User(authid.clone()));
+                    }
                     rpcenv.set_auth_id(Some(authid));
                     user_info = info;
                 }
                 Err(auth_err) => {
+                    #[cfg(feature = "rate-limited-stream")]
+                    if let Some(handle) = rate_limit_tags.as_ref() {
+                        handle.lock().unwrap().clear();
+                    }
                     let err = match auth_err {
                         AuthError::Generic(err) => err,
                         AuthError::NoData => {
@@ -1045,6 +1153,11 @@ impl Formatted {
                     return Err(err);
                 }
             }
+        } else {
+            #[cfg(feature = "rate-limited-stream")]
+            if let Some(handle) = rate_limit_tags.as_ref() {
+                handle.lock().unwrap().clear();
+            }
         }
 
         match api_method {
@@ -1108,6 +1221,8 @@ impl Unformatted {
             full_path,
             relative_path_components,
             mut rpcenv,
+            #[cfg(feature = "rate-limited-stream")]
+            rate_limit_tags,
         }: ApiRequestData<'_>,
     ) -> Result<Response<Body>, Error> {
         if relative_path_components.is_empty() {
@@ -1133,10 +1248,20 @@ impl Unformatted {
         if auth_required {
             match config.check_auth(&parts.headers, &parts.method).await {
                 Ok((authid, info)) => {
+                    #[cfg(feature = "rate-limited-stream")]
+                    if let Some(handle) = rate_limit_tags.as_ref() {
+                        let mut guard = handle.lock().unwrap();
+                        guard.clear();
+                        guard.push(RateLimiterTag::User(authid.clone()));
+                    }
                     rpcenv.set_auth_id(Some(authid));
                     user_info = info;
                 }
                 Err(auth_err) => {
+                    #[cfg(feature = "rate-limited-stream")]
+                    if let Some(handle) = rate_limit_tags.as_ref() {
+                        handle.lock().unwrap().clear();
+                    }
                     let err = match auth_err {
                         AuthError::Generic(err) => err,
                         AuthError::NoData => {
@@ -1153,6 +1278,10 @@ impl Unformatted {
                 }
             }
         } else {
+            #[cfg(feature = "rate-limited-stream")]
+            if let Some(handle) = rate_limit_tags.as_ref() {
+                handle.lock().unwrap().clear();
+            }
             user_info = Box::new(EmptyUserInformation {});
         }
 
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup v3 1/3] api: taffic-control: update/delete users on rule correctly

[Hannes Laimer]

Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/api2/config/traffic_control.rs | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/src/api2/config/traffic_control.rs b/src/api2/config/traffic_control.rs
index e02aa20a..49b1267e 100644
--- a/src/api2/config/traffic_control.rs
+++ b/src/api2/config/traffic_control.rs
@@ -116,6 +116,8 @@ pub enum DeletableProperty {
     Comment,
     /// Delete the timeframe property
     Timeframe,
+    /// Delete the users property
+    Users,
 }
 
 // fixme: use  TrafficControlUpdater
@@ -187,6 +189,9 @@ pub fn update_traffic_control(
                 DeletableProperty::Timeframe => {
                     data.timeframe = None;
                 }
+                DeletableProperty::Users => {
+                    data.users = None;
+                }
             }
         }
     }
@@ -222,6 +227,9 @@ pub fn update_traffic_control(
     if update.timeframe.is_some() {
         data.timeframe = update.timeframe;
     }
+    if update.users.is_some() {
+        data.users = update.users;
+    }
 
     config.set_data(&name, "rule", &data)?;
 
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup v3 2/3] traffic-control: handle users specified in a rule correctly

[Hannes Laimer]

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/bin/proxmox-backup-proxy.rs |  12 +++-
 src/traffic_control_cache.rs    | 105 ++++++++++++++++++++++++++++----
 2 files changed, 103 insertions(+), 14 deletions(-)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index cfd93f92..92a8cb3c 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -17,6 +17,7 @@ use openssl::ssl::SslAcceptor;
 use serde_json::{json, Value};
 
 use proxmox_http::Body;
+use proxmox_http::RateLimiterTag;
 use proxmox_lang::try_block;
 use proxmox_router::{RpcEnvironment, RpcEnvironmentType};
 use proxmox_sys::fs::CreateOptions;
@@ -955,6 +956,7 @@ async fn run_traffic_control_updater() {
 
 fn lookup_rate_limiter(
     peer: std::net::SocketAddr,
+    tags: &[RateLimiterTag],
 ) -> (Option<SharedRateLimit>, Option<SharedRateLimit>) {
     let mut cache = TRAFFIC_CONTROL_CACHE.lock().unwrap();
 
@@ -962,7 +964,15 @@ fn lookup_rate_limiter(
 
     cache.reload(now);
 
-    let (_rule_name, read_limiter, write_limiter) = cache.lookup_rate_limiter(peer, now);
+    let user = tags.iter().find_map(|tag| match tag {
+        RateLimiterTag::User(user) => Some(user.as_str()),
+    });
+
+    let authid = user.and_then(|s| s.parse::<pbs_api_types::Authid>().ok());
+    let user_parsed = authid.as_ref().map(|auth_id| auth_id.user());
+
+    let (_rule_name, read_limiter, write_limiter) =
+        cache.lookup_rate_limiter(peer, now, user_parsed);
 
     (read_limiter, write_limiter)
 }
diff --git a/src/traffic_control_cache.rs b/src/traffic_control_cache.rs
index 830a8c04..95a7bf0b 100644
--- a/src/traffic_control_cache.rs
+++ b/src/traffic_control_cache.rs
@@ -13,7 +13,7 @@ use proxmox_section_config::SectionConfigData;
 
 use proxmox_time::{parse_daily_duration, DailyDuration, TmEditor};
 
-use pbs_api_types::TrafficControlRule;
+use pbs_api_types::{TrafficControlRule, Userid};
 
 use pbs_config::ConfigVersionCache;
 
@@ -31,6 +31,12 @@ struct ParsedTcRule {
     timeframe: Vec<DailyDuration>, // parsed timeframe
 }
 
+/// Represents a matched traffic control rule with its network prefix length and user rule status.
+///
+/// Used during rule lookup to track the best matching rule, where user-specific rules
+/// take precedence over IP-only rules, and longer network prefixes are preferred.
+type RuleMatch<'a> = (&'a ParsedTcRule, u8, bool); // (rule, network_prefix_length, is_user_rule)
+
 /// Traffic control statistics
 pub struct TrafficStat {
     /// Total incoming traffic (bytes)
@@ -322,6 +328,7 @@ impl TrafficControlCache {
     ///
     /// - Rules where timeframe does not match are skipped.
     /// - Rules with smaller network size have higher priority.
+    /// - Rules with users are prioritized over IP-only rules, if multiple match
     ///
     /// Behavior is undefined if more than one rule matches after
     /// above selection.
@@ -329,10 +336,14 @@ impl TrafficControlCache {
         &self,
         peer: SocketAddr,
         now: i64,
+        user: Option<&Userid>,
     ) -> (&str, Option<SharedRateLimit>, Option<SharedRateLimit>) {
         let peer_ip = canonical_ip(peer.ip());
 
-        log::debug!("lookup_rate_limiter: {:?}", peer_ip);
+        log::debug!(
+            "lookup_rate_limiter: {peer_ip:?} - {}",
+            user.map_or("(no user)", |u| u.as_str())
+        );
 
         let now = match TmEditor::with_epoch(now, self.use_utc) {
             Ok(now) => now,
@@ -342,19 +353,33 @@ impl TrafficControlCache {
             }
         };
 
-        let mut last_rule_match = None;
+        let mut last_rule_match: Option<RuleMatch> = None;
 
         for rule in self.rules.iter() {
             if !timeframe_match(&rule.timeframe, &now) {
                 continue;
             }
 
+            if let Some(ref rule_users) = rule.config.users {
+                if let Some(cur_user) = user {
+                    if !rule_users.iter().any(|u| u == cur_user) {
+                        continue;
+                    }
+                } else {
+                    continue;
+                }
+            }
+
             if let Some(match_len) = network_match_len(&rule.networks, &peer_ip) {
+                let is_user_rule = rule.config.users.is_some();
                 match last_rule_match {
-                    None => last_rule_match = Some((rule, match_len)),
-                    Some((_, last_len)) => {
-                        if match_len > last_len {
-                            last_rule_match = Some((rule, match_len));
+                    None => last_rule_match = Some((rule, match_len, is_user_rule)),
+                    Some((_, last_len, last_is_user)) => {
+                        // Prefer rules with users over IP-only rules; for same class use longest prefix
+                        if (is_user_rule && !last_is_user)
+                            || (is_user_rule == last_is_user && match_len > last_len)
+                        {
+                            last_rule_match = Some((rule, match_len, is_user_rule));
                         }
                     }
                 }
@@ -362,7 +387,7 @@ impl TrafficControlCache {
         }
 
         match last_rule_match {
-            Some((rule, _)) => {
+            Some((rule, _, _)) => {
                 match self.limiter_map.get(&rule.config.name) {
                     Some((read_limiter, write_limiter)) => (
                         &rule.config.name,
@@ -454,34 +479,88 @@ rule: somewhere
         let somewhere = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 1234);
 
         let (rule, read_limiter, write_limiter) =
-            cache.lookup_rate_limiter(somewhere, THURSDAY_80_00);
+            cache.lookup_rate_limiter(somewhere, THURSDAY_80_00, None);
         assert_eq!(rule, "somewhere");
         assert!(read_limiter.is_some());
         assert!(write_limiter.is_some());
 
-        let (rule, read_limiter, write_limiter) = cache.lookup_rate_limiter(local, THURSDAY_19_00);
+        let (rule, read_limiter, write_limiter) =
+            cache.lookup_rate_limiter(local, THURSDAY_19_00, None);
         assert_eq!(rule, "rule2");
         assert!(read_limiter.is_some());
         assert!(write_limiter.is_some());
 
         let (rule, read_limiter, write_limiter) =
-            cache.lookup_rate_limiter(gateway, THURSDAY_15_00);
+            cache.lookup_rate_limiter(gateway, THURSDAY_15_00, None);
         assert_eq!(rule, "rule1");
         assert!(read_limiter.is_some());
         assert!(write_limiter.is_some());
 
         let (rule, read_limiter, write_limiter) =
-            cache.lookup_rate_limiter(gateway, THURSDAY_19_00);
+            cache.lookup_rate_limiter(gateway, THURSDAY_19_00, None);
         assert_eq!(rule, "somewhere");
         assert!(read_limiter.is_some());
         assert!(write_limiter.is_some());
 
         let (rule, read_limiter, write_limiter) =
-            cache.lookup_rate_limiter(private, THURSDAY_19_00);
+            cache.lookup_rate_limiter(private, THURSDAY_19_00, None);
         assert_eq!(rule, "rule2");
         assert!(read_limiter.is_some());
         assert!(write_limiter.is_some());
 
         Ok(())
     }
+
+    #[test]
+    fn test_user_based_rule_match_and_precedence() -> Result<(), Error> {
+        // rule user1: user-specific for alice@pam on 192.168.2.0/24
+        // rule ip1: ip-only broader network also matches, but user rule should win
+        // rule user2: user-specific for bob@pam but different network shouldn't match
+        let config_data = "
+rule: user1
+	comment user rule for alice
+	network 192.168.2.0/24
+	rate-in 50000000
+	rate-out 50000000
+	users alice@pam
+
+rule: ip1
+	network 192.168.2.0/24
+	rate-in 100000000
+	rate-out 100000000
+
+rule: user2
+	network 10.0.0.0/8
+	rate-in 75000000
+	rate-out 75000000
+	users bob@pam
+";
+
+        let config = pbs_config::traffic_control::CONFIG.parse("testconfig", config_data)?;
+
+        let mut cache = TrafficControlCache::new();
+        cache.use_utc = true;
+        cache.use_shared_memory = false; // avoid permission problems in test environment
+
+        cache.update_config(&config)?;
+
+        const NOW: i64 = make_test_time(0, 12, 0);
+        let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 2, 55)), 1234);
+
+        // No user -> should match ip1
+        let (rule, _, _) = cache.lookup_rate_limiter(peer, NOW, None);
+        assert_eq!(rule, "ip1");
+
+        // alice@pam -> should match user1 and take precedence over ip1
+        let alice = Userid::try_from("alice@pam".to_string())?;
+        let (rule, _, _) = cache.lookup_rate_limiter(peer, NOW, Some(&alice));
+        assert_eq!(rule, "user1");
+
+        // bob@pam on same peer/network -> user2 is different network, should fall back to ip1
+        let bob = Userid::try_from("bob@pam".to_string())?;
+        let (rule, _, _) = cache.lookup_rate_limiter(peer, NOW, Some(&bob));
+        assert_eq!(rule, "ip1");
+
+        Ok(())
+    }
 }
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup v3 3/3] ui: traffic-control: add users field in edit form and list

[Hannes Laimer]

Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 www/config/TrafficControlView.js |  7 +++++++
 www/window/TrafficControlEdit.js | 18 ++++++++++++++++++
 2 files changed, 25 insertions(+)

diff --git a/www/config/TrafficControlView.js b/www/config/TrafficControlView.js
index 0b22d29a..5cfec82b 100644
--- a/www/config/TrafficControlView.js
+++ b/www/config/TrafficControlView.js
@@ -181,6 +181,13 @@ Ext.define('PBS.config.TrafficControlView', {
             renderer: 'render_bandwidth',
             dataIndex: 'burst-out',
         },
+        {
+            header: gettext('Users'),
+            flex: 3,
+            sortable: true,
+            renderer: (users) => (users ? Ext.String.htmlEncode(users.join(', ')) : ''),
+            dataIndex: 'users',
+        },
         {
             header: gettext('Networks'),
             flex: 3,
diff --git a/www/window/TrafficControlEdit.js b/www/window/TrafficControlEdit.js
index 0bbbf363..2063c107 100644
--- a/www/window/TrafficControlEdit.js
+++ b/www/window/TrafficControlEdit.js
@@ -215,6 +215,7 @@ Ext.define('PBS.window.TrafficControlEdit', {
                 PBS.Utils.delete_if_default(values, 'rate-out');
                 PBS.Utils.delete_if_default(values, 'burst-in');
                 PBS.Utils.delete_if_default(values, 'burst-out');
+                PBS.Utils.delete_if_default(values, 'users');
                 if (typeof values.delete === 'string') {
                     values.delete = values.delete.split(',');
                 }
@@ -276,6 +277,23 @@ Ext.define('PBS.window.TrafficControlEdit', {
         ],
 
         columnB: [
+            {
+                xtype: 'pmxUserSelector',
+                fieldLabel: gettext('Users'),
+                name: 'users',
+                multiSelect: true,
+                allowBlank: true,
+                cbind: {
+                    deleteEmpty: '{!isCreate}',
+                },
+                emptyText: gettext('Applies to all users'),
+                autoEl: {
+                    tag: 'div',
+                    'data-qtip': gettext(
+                        'Limit applies only to authenticated requests by these users. Overrides IP-only rules when both match. If networks are specified on this rule as well, it\'ll only apply if the users request comes from one of the specified networks.',
+                    ),
+                },
+            },
             {
                 xtype: 'proxmoxtextfield',
                 fieldLabel: gettext('Network(s)'),
-- 
2.47.3



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox v3 1/3] pbs-api-types: allow traffic-control rules to match users

[Christian Ebner]

one nit inline I missed last time, otherwise LGTM!

On 11/10/25 2:43 PM, Hannes Laimer wrote:
> Extend traffic-control rules with an optional list of user IDs so
> API traffic can be limited per user in addition to IP-based rules.
> 
> Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
> Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
> ---
>   pbs-api-types/src/traffic_control.rs | 9 +++++++++
>   1 file changed, 9 insertions(+)
> 
> diff --git a/pbs-api-types/src/traffic_control.rs b/pbs-api-types/src/traffic_control.rs
> index 2a359eda..12fc8c93 100644
> --- a/pbs-api-types/src/traffic_control.rs
> +++ b/pbs-api-types/src/traffic_control.rs
> @@ -6,6 +6,7 @@ use proxmox_schema::{api, ApiType, Schema, StringSchema, Updater};
>   use proxmox_schema::api_types::CIDR_SCHEMA;
>   
>   use crate::{DAILY_DURATION_FORMAT, PROXMOX_SAFE_ID_FORMAT, SINGLE_LINE_COMMENT_SCHEMA};
> +use crate::Userid;

nit: cargo fmt would place this above the schema and format imports.

>   
>   pub const TRAFFIC_CONTROL_TIMEFRAME_SCHEMA: Schema =
>       StringSchema::new("Timeframe to specify when the rule is active.")
> @@ -125,6 +126,11 @@ pub struct ClientRateLimitConfig {
>               },
>               optional: true,
>           },
> +        users: {
> +            type: Array,
> +            items: { type: Userid },
> +            optional: true,
> +        },
>       },
>   )]
>   #[derive(Clone, Serialize, Deserialize, PartialEq, Updater)]
> @@ -146,6 +152,9 @@ pub struct TrafficControlRule {
>       /// Enable the rule at specific times
>       #[serde(skip_serializing_if = "Option::is_none")]
>       pub timeframe: Option<Vec<String>>,
> +    /// Rule applies to authenticated API requests of any of these users (overrides IP-only rules)
> +    #[serde(skip_serializing_if = "Option::is_none")]
> +    pub users: Option<Vec<Userid>>,
>   }
>   
>   #[api(



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox v3 2/3] http: track user tag updates on rate-limited streams

[Christian Ebner]

nit: needs reformatting via cargo fmt

On 11/10/25 2:42 PM, Hannes Laimer wrote:
> Introduce rate-limit tags with a user variant and let rate-limited
> streams hold a shared handle so callbacks can refresh limits
> whenever the tag set changes.
> 
> If we decide to implement something like [1] in the future this could
> potentially include group rate-limits for example.
> 
> [1] https://bugzilla.proxmox.com/show_bug.cgi?id=5867
> 
> Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
> ---
>   proxmox-http/src/lib.rs                 |  2 +-
>   proxmox-http/src/rate_limited_stream.rs | 40 +++++++++++++++++++++----
>   2 files changed, 36 insertions(+), 6 deletions(-)
> 
> diff --git a/proxmox-http/src/lib.rs b/proxmox-http/src/lib.rs
> index 8b6953b0..990f8f36 100644
> --- a/proxmox-http/src/lib.rs
> +++ b/proxmox-http/src/lib.rs
> @@ -34,7 +34,7 @@ pub use rate_limiter::{RateLimit, RateLimiter, RateLimiterVec, ShareableRateLimi
>   #[cfg(feature = "rate-limited-stream")]
>   mod rate_limited_stream;
>   #[cfg(feature = "rate-limited-stream")]
> -pub use rate_limited_stream::RateLimitedStream;
> +pub use rate_limited_stream::{RateLimitedStream, RateLimiterTag, RateLimiterTags};
>   
>   #[cfg(feature = "body")]
>   mod body;
> diff --git a/proxmox-http/src/rate_limited_stream.rs b/proxmox-http/src/rate_limited_stream.rs
> index e24df7af..6b525591 100644
> --- a/proxmox-http/src/rate_limited_stream.rs
> +++ b/proxmox-http/src/rate_limited_stream.rs
> @@ -15,8 +15,15 @@ use super::{RateLimiter, ShareableRateLimit};
>   
>   type SharedRateLimit = Arc<dyn ShareableRateLimit>;
>   
> +#[derive(Clone, Debug, PartialEq, Eq)]
> +pub enum RateLimiterTag {
> +    User(String),
> +}
> +
> +pub type RateLimiterTags = Vec<RateLimiterTag>;
> +
>   pub type RateLimiterCallback =
> -    dyn Fn() -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send;
> +    dyn Fn(&[RateLimiterTag]) -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send;
>   
>   /// A rate limited stream using [RateLimiter]
>   pub struct RateLimitedStream<S> {
> @@ -26,6 +33,8 @@ pub struct RateLimitedStream<S> {
>       write_delay: Option<Pin<Box<Sleep>>>,
>       update_limiter_cb: Option<Box<RateLimiterCallback>>,
>       last_limiter_update: Instant,
> +    tag_handle: Option<Arc<Mutex<RateLimiterTags>>>,
> +    last_tags: Option<RateLimiterTags>,
>       stream: S,
>   }
>   
> @@ -53,6 +62,8 @@ impl<S> RateLimitedStream<S> {
>               write_delay: None,
>               update_limiter_cb: None,
>               last_limiter_update: Instant::now(),
> +            tag_handle: None,
> +            last_tags: None,
>               stream,
>           }
>       }
> @@ -64,12 +75,13 @@ impl<S> RateLimitedStream<S> {
>       /// Note: This function is called within an async context, so it
>       /// should be fast and must not block.
>       pub fn with_limiter_update_cb<
> -        F: Fn() -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send + 'static,
> +        F: Fn(&[RateLimiterTag]) -> (Option<SharedRateLimit>, Option<SharedRateLimit>) + Send + 'static,
>       >(
>           stream: S,
>           update_limiter_cb: F,
>       ) -> Self {
> -        let (read_limiter, write_limiter) = update_limiter_cb();
> +        let tag_handle = Some(Arc::new(Mutex::new(Vec::new())));
> +        let (read_limiter, write_limiter) = update_limiter_cb(&[]);
>           Self {
>               read_limiter,
>               read_delay: None,
> @@ -77,15 +89,29 @@ impl<S> RateLimitedStream<S> {
>               write_delay: None,
>               update_limiter_cb: Some(Box::new(update_limiter_cb)),
>               last_limiter_update: Instant::now(),
> +            tag_handle,
> +            last_tags: None,
>               stream,
>           }
>       }
>   
>       fn update_limiters(&mut self) {
>           if let Some(ref update_limiter_cb) = self.update_limiter_cb {
> -            if self.last_limiter_update.elapsed().as_secs() >= 5 {
> +            let mut force_update = false;
> +            let current_tags = self
> +                .tag_handle
> +                .as_ref()
> +                .map(|handle| handle.lock().unwrap().clone());
> +
> +            if self.last_tags != current_tags {
> +                self.last_tags = current_tags.clone();
> +                force_update = true;
> +            }
> +
> +            if force_update || self.last_limiter_update.elapsed().as_secs() >= 5 {
>                   self.last_limiter_update = Instant::now();
> -                let (read_limiter, write_limiter) = update_limiter_cb();
> +                let tags = self.last_tags.as_ref().map(|tags| tags.as_slice()).unwrap_or(&[]);
> +                let (read_limiter, write_limiter) = update_limiter_cb(tags);
>                   self.read_limiter = read_limiter;
>                   self.write_limiter = write_limiter;
>               }
> @@ -99,6 +125,10 @@ impl<S> RateLimitedStream<S> {
>       pub fn inner_mut(&mut self) -> &mut S {
>           &mut self.stream
>       }
> +
> +    pub fn tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
> +        self.tag_handle.as_ref().map(Arc::clone)
> +    }
>   }
>   
>   fn register_traffic(limiter: &dyn ShareableRateLimit, count: usize) -> Option<Pin<Box<Sleep>>> {



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox v3 3/3] rest-server: propagate rate-limit tags from authenticated users

[Christian Ebner]

nit: needs reformatting via cargo fmt

On 11/10/25 2:43 PM, Hannes Laimer wrote:
> Tie REST connections rate-limiter callbacks to a shared tag handle
> so we can push authenticated user IDs into the limiter, keeping
> tags in sync whenever we clear or update them.
> 
> Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
> ---
>   proxmox-rest-server/src/connection.rs |  11 ++-
>   proxmox-rest-server/src/rest.rs       | 137 +++++++++++++++++++++++++-
>   2 files changed, 141 insertions(+), 7 deletions(-)
> 
> diff --git a/proxmox-rest-server/src/connection.rs b/proxmox-rest-server/src/connection.rs
> index 9511b7cb..ff2ee139 100644
> --- a/proxmox-rest-server/src/connection.rs
> +++ b/proxmox-rest-server/src/connection.rs
> @@ -24,7 +24,7 @@ use tokio_openssl::SslStream;
>   use tokio_stream::wrappers::ReceiverStream;
>   
>   #[cfg(feature = "rate-limited-stream")]
> -use proxmox_http::{RateLimitedStream, ShareableRateLimit};
> +use proxmox_http::{RateLimitedStream, RateLimiterTag, ShareableRateLimit};
>   
>   #[cfg(feature = "rate-limited-stream")]
>   pub type SharedRateLimit = Arc<dyn ShareableRateLimit>;
> @@ -165,7 +165,10 @@ type InsecureClientStreamResult = Pin<Box<InsecureClientStream>>;
>   type ClientStreamResult = Pin<Box<SslStream<InsecureClientStream>>>;
>   
>   #[cfg(feature = "rate-limited-stream")]
> -type LookupRateLimiter = dyn Fn(std::net::SocketAddr) -> (Option<SharedRateLimit>, Option<SharedRateLimit>)
> +type LookupRateLimiter = dyn Fn(
> +        std::net::SocketAddr,
> +        &[RateLimiterTag],
> +    ) -> (Option<SharedRateLimit>, Option<SharedRateLimit>)
>       + Send
>       + Sync
>       + 'static;
> @@ -369,7 +372,9 @@ impl AcceptBuilder {
>   
>           #[cfg(feature = "rate-limited-stream")]
>           let socket = match self.lookup_rate_limiter.clone() {
> -            Some(lookup) => RateLimitedStream::with_limiter_update_cb(socket, move || lookup(peer)),
> +            Some(lookup) => RateLimitedStream::with_limiter_update_cb(socket, move |tags| {
> +                lookup(peer, tags)
> +            }),
>               None => RateLimitedStream::with_limiter(socket, None, None),
>           };
>   
> diff --git a/proxmox-rest-server/src/rest.rs b/proxmox-rest-server/src/rest.rs
> index b76c4bc9..6e9692ae 100644
> --- a/proxmox-rest-server/src/rest.rs
> +++ b/proxmox-rest-server/src/rest.rs
> @@ -29,6 +29,10 @@ use tower_service::Service;
>   use url::form_urlencoded;
>   
>   use proxmox_http::Body;
> +#[cfg(feature = "rate-limited-stream")]
> +use proxmox_http::{RateLimiterTag, RateLimiterTags};
> +#[cfg(not(feature = "rate-limited-stream"))]
> +type RateLimiterTags = ();
>   use proxmox_router::{
>       check_api_permission, ApiHandler, ApiMethod, HttpError, Permission, RpcEnvironment,
>       RpcEnvironmentType, UserInformation,
> @@ -86,10 +90,26 @@ impl RestServer {
>           }
>       }
>   
> -    pub fn api_service(&self, peer: &dyn PeerAddress) -> Result<ApiService, Error> {
> +    #[cfg(not(feature = "rate-limited-stream"))]
> +    pub fn api_service<T>(&self, peer: &T) -> Result<ApiService, Error>
> +    where
> +        T: PeerAddress + ?Sized,
> +    {
> +        Ok(ApiService {
> +            peer: peer.peer_addr()?,
> +            api_config: Arc::clone(&self.api_config),
> +        })
> +    }
> +
> +    #[cfg(feature = "rate-limited-stream")]
> +    pub fn api_service<T>(&self, peer: &T) -> Result<ApiService, Error>
> +    where
> +        T: PeerAddress + PeerRateLimitTags + ?Sized,
> +    {
>           Ok(ApiService {
>               peer: peer.peer_addr()?,
>               api_config: Arc::clone(&self.api_config),
> +            rate_limit_tags: peer.rate_limiter_tag_handle(),
>           })
>       }
>   }
> @@ -185,6 +205,11 @@ pub trait PeerAddress {
>       fn peer_addr(&self) -> Result<std::net::SocketAddr, Error>;
>   }
>   
> +#[cfg(feature = "rate-limited-stream")]
> +pub trait PeerRateLimitTags {
> +    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>>;
> +}
> +
>   // tokio_openssl's SslStream requires the stream to be pinned in order to accept it, and we need to
>   // accept before the peer address is requested, so let's just generally implement this for
>   // Pin<Box<T>>
> @@ -221,6 +246,41 @@ impl<T: PeerAddress> PeerAddress for proxmox_http::RateLimitedStream<T> {
>       }
>   }
>   
> +#[cfg(feature = "rate-limited-stream")]
> +impl<T: PeerRateLimitTags> PeerRateLimitTags for Pin<Box<T>> {
> +    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
> +        T::rate_limiter_tag_handle(&**self)
> +    }
> +}
> +
> +#[cfg(feature = "rate-limited-stream")]
> +impl<T: PeerRateLimitTags> PeerRateLimitTags for tokio_openssl::SslStream<T> {
> +    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
> +        self.get_ref().rate_limiter_tag_handle()
> +    }
> +}
> +
> +#[cfg(feature = "rate-limited-stream")]
> +impl PeerRateLimitTags for tokio::net::TcpStream {
> +    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
> +        None
> +    }
> +}
> +
> +#[cfg(feature = "rate-limited-stream")]
> +impl PeerRateLimitTags for tokio::net::UnixStream {
> +    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
> +        None
> +    }
> +}
> +
> +#[cfg(feature = "rate-limited-stream")]
> +impl<T> PeerRateLimitTags for proxmox_http::RateLimitedStream<T> {
> +    fn rate_limiter_tag_handle(&self) -> Option<Arc<Mutex<RateLimiterTags>>> {
> +        self.tag_handle()
> +    }
> +}
> +
>   // Helper [Service] containing the peer Address
>   //
>   // The lower level connection [Service] implementation on
> @@ -233,6 +293,8 @@ impl<T: PeerAddress> PeerAddress for proxmox_http::RateLimitedStream<T> {
>   pub struct ApiService {
>       pub peer: std::net::SocketAddr,
>       pub api_config: Arc<ApiConfig>,
> +    #[cfg(feature = "rate-limited-stream")]
> +    pub rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>>,
>   }
>   
>   impl ApiService {
> @@ -354,6 +416,10 @@ impl Service<Request<Incoming>> for ApiService {
>               Some(proxied_peer) => proxied_peer,
>               None => self.peer,
>           };
> +        #[cfg(feature = "rate-limited-stream")]
> +        let rate_limit_tags = self.rate_limit_tags.clone();
> +        #[cfg(not(feature = "rate-limited-stream"))]
> +        let rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>> = None;
>   
>           let header = self.api_config
>               .auth_cookie_name
> @@ -368,7 +434,15 @@ impl Service<Request<Incoming>> for ApiService {
>                });
>   
>           async move {
> -            let mut response = match Arc::clone(&config).handle_request(req, &peer).await {
> +            #[cfg(feature = "rate-limited-stream")]
> +            if let Some(handle) = rate_limit_tags.as_ref() {
> +                handle.lock().unwrap().clear();
> +            }
> +
> +            let mut response = match Arc::clone(&config)
> +                .handle_request(req, &peer, rate_limit_tags.clone())
> +                .await
> +            {
>                   Ok(response) => response,
>                   Err(err) => {
>                       let (err, code) = match err.downcast_ref::<HttpError>() {
> @@ -860,6 +934,8 @@ impl ApiConfig {
>           self: Arc<ApiConfig>,
>           req: Request<Incoming>,
>           peer: &std::net::SocketAddr,
> +        #[cfg_attr(not(feature = "rate-limited-stream"), allow(unused_variables))]
> +        rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>>,
>       ) -> Result<Response<Body>, Error> {
>           let (parts, body) = req.into_parts();
>           let method = parts.method.clone();
> @@ -890,6 +966,8 @@ impl ApiConfig {
>                       full_path: &path,
>                       relative_path_components,
>                       rpcenv,
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    rate_limit_tags: rate_limit_tags.clone(),
>                   })
>                   .await;
>           }
> @@ -901,13 +979,29 @@ impl ApiConfig {
>           if components.is_empty() {
>               match self.check_auth(&parts.headers, &method).await {
>                   Ok((auth_id, _user_info)) => {
> -                    rpcenv.set_auth_id(Some(auth_id));
> +                    rpcenv.set_auth_id(Some(auth_id.clone()));
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    if let Some(handle) = rate_limit_tags.as_ref() {
> +                        let mut guard = handle.lock().unwrap();
> +                        guard.clear();
> +                        guard.push(RateLimiterTag::User(auth_id));
> +                    }
>                       return Ok(self.get_index(rpcenv, parts).await);
>                   }
>                   Err(AuthError::Generic(_)) => {
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    if let Some(handle) = rate_limit_tags.as_ref() {
> +                        handle.lock().unwrap().clear();
> +                    }
>                       tokio::time::sleep_until(Instant::from_std(delay_unauth_time())).await;
>                   }
> -                Err(AuthError::NoData) => {}
> +                Err(AuthError::NoData) =>
> +                {
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    if let Some(handle) = rate_limit_tags.as_ref() {
> +                        handle.lock().unwrap().clear();
> +                    }
> +                }
>               }
>               Ok(self.get_index(rpcenv, parts).await)
>           } else {
> @@ -975,6 +1069,8 @@ pub struct ApiRequestData<'a> {
>       full_path: &'a str,
>       relative_path_components: &'a [&'a str],
>       rpcenv: RestEnvironment,
> +    #[cfg(feature = "rate-limited-stream")]
> +    rate_limit_tags: Option<Arc<Mutex<RateLimiterTags>>>,
>   }
>   
>   pub(crate) struct Formatted {
> @@ -992,6 +1088,8 @@ impl Formatted {
>               full_path,
>               relative_path_components,
>               mut rpcenv,
> +            #[cfg(feature = "rate-limited-stream")]
> +            rate_limit_tags,
>           }: ApiRequestData<'_>,
>       ) -> Result<Response<Body>, Error> {
>           if relative_path_components.is_empty() {
> @@ -1026,10 +1124,20 @@ impl Formatted {
>           if auth_required {
>               match config.check_auth(&parts.headers, &parts.method).await {
>                   Ok((authid, info)) => {
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    if let Some(handle) = rate_limit_tags.as_ref() {
> +                        let mut guard = handle.lock().unwrap();
> +                        guard.clear();
> +                        guard.push(RateLimiterTag::User(authid.clone()));
> +                    }
>                       rpcenv.set_auth_id(Some(authid));
>                       user_info = info;
>                   }
>                   Err(auth_err) => {
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    if let Some(handle) = rate_limit_tags.as_ref() {
> +                        handle.lock().unwrap().clear();
> +                    }
>                       let err = match auth_err {
>                           AuthError::Generic(err) => err,
>                           AuthError::NoData => {
> @@ -1045,6 +1153,11 @@ impl Formatted {
>                       return Err(err);
>                   }
>               }
> +        } else {
> +            #[cfg(feature = "rate-limited-stream")]
> +            if let Some(handle) = rate_limit_tags.as_ref() {
> +                handle.lock().unwrap().clear();
> +            }
>           }
>   
>           match api_method {
> @@ -1108,6 +1221,8 @@ impl Unformatted {
>               full_path,
>               relative_path_components,
>               mut rpcenv,
> +            #[cfg(feature = "rate-limited-stream")]
> +            rate_limit_tags,
>           }: ApiRequestData<'_>,
>       ) -> Result<Response<Body>, Error> {
>           if relative_path_components.is_empty() {
> @@ -1133,10 +1248,20 @@ impl Unformatted {
>           if auth_required {
>               match config.check_auth(&parts.headers, &parts.method).await {
>                   Ok((authid, info)) => {
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    if let Some(handle) = rate_limit_tags.as_ref() {
> +                        let mut guard = handle.lock().unwrap();
> +                        guard.clear();
> +                        guard.push(RateLimiterTag::User(authid.clone()));
> +                    }
>                       rpcenv.set_auth_id(Some(authid));
>                       user_info = info;
>                   }
>                   Err(auth_err) => {
> +                    #[cfg(feature = "rate-limited-stream")]
> +                    if let Some(handle) = rate_limit_tags.as_ref() {
> +                        handle.lock().unwrap().clear();
> +                    }
>                       let err = match auth_err {
>                           AuthError::Generic(err) => err,
>                           AuthError::NoData => {
> @@ -1153,6 +1278,10 @@ impl Unformatted {
>                   }
>               }
>           } else {
> +            #[cfg(feature = "rate-limited-stream")]
> +            if let Some(handle) = rate_limit_tags.as_ref() {
> +                handle.lock().unwrap().clear();
> +            }
>               user_info = Box::new(EmptyUserInformation {});
>           }
>   



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup v3 3/3] ui: traffic-control: add users field in edit form and list

[Christian Ebner]

On 11/10/25 2:42 PM, Hannes Laimer wrote:
> Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
> Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>

nit: incorrect order of tags

> ---
>   www/config/TrafficControlView.js |  7 +++++++
>   www/window/TrafficControlEdit.js | 18 ++++++++++++++++++
>   2 files changed, 25 insertions(+)
> 
> diff --git a/www/config/TrafficControlView.js b/www/config/TrafficControlView.js
> index 0b22d29a..5cfec82b 100644
> --- a/www/config/TrafficControlView.js
> +++ b/www/config/TrafficControlView.js
> @@ -181,6 +181,13 @@ Ext.define('PBS.config.TrafficControlView', {
>               renderer: 'render_bandwidth',
>               dataIndex: 'burst-out',
>           },
> +        {
> +            header: gettext('Users'),
> +            flex: 3,
> +            sortable: true,
> +            renderer: (users) => (users ? Ext.String.htmlEncode(users.join(', ')) : ''),
> +            dataIndex: 'users',
> +        },
>           {
>               header: gettext('Networks'),
>               flex: 3,
> diff --git a/www/window/TrafficControlEdit.js b/www/window/TrafficControlEdit.js
> index 0bbbf363..2063c107 100644
> --- a/www/window/TrafficControlEdit.js
> +++ b/www/window/TrafficControlEdit.js
> @@ -215,6 +215,7 @@ Ext.define('PBS.window.TrafficControlEdit', {
>                   PBS.Utils.delete_if_default(values, 'rate-out');
>                   PBS.Utils.delete_if_default(values, 'burst-in');
>                   PBS.Utils.delete_if_default(values, 'burst-out');
> +                PBS.Utils.delete_if_default(values, 'users');
>                   if (typeof values.delete === 'string') {
>                       values.delete = values.delete.split(',');
>                   }
> @@ -276,6 +277,23 @@ Ext.define('PBS.window.TrafficControlEdit', {
>           ],
>   
>           columnB: [
> +            {
> +                xtype: 'pmxUserSelector',
> +                fieldLabel: gettext('Users'),
> +                name: 'users',
> +                multiSelect: true,
> +                allowBlank: true,
> +                cbind: {
> +                    deleteEmpty: '{!isCreate}',
> +                },
> +                emptyText: gettext('Applies to all users'),
> +                autoEl: {
> +                    tag: 'div',
> +                    'data-qtip': gettext(
> +                        'Limit applies only to authenticated requests by these users. Overrides IP-only rules when both match. If networks are specified on this rule as well, it\'ll only apply if the users request comes from one of the specified networks.',

nit: proxmox-biome would reformat this to use double quotes for the 
string delimiter, so the singe quote does not need to be escaped.

> +                    ),
> +                },
> +            },
>               {
>                   xtype: 'proxmoxtextfield',
>                   fieldLabel: gettext('Network(s)'),



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup v3 2/3] traffic-control: handle users specified in a rule correctly

[Christian Ebner]

This patch warrants a commit message IMO, as it touches quite a bit of 
traffic rule matching logic.

On 11/10/25 2:43 PM, Hannes Laimer wrote:
> Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
> ---
>   src/bin/proxmox-backup-proxy.rs |  12 +++-
>   src/traffic_control_cache.rs    | 105 ++++++++++++++++++++++++++++----
>   2 files changed, 103 insertions(+), 14 deletions(-)
> 
> diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
> index cfd93f92..92a8cb3c 100644
> --- a/src/bin/proxmox-backup-proxy.rs
> +++ b/src/bin/proxmox-backup-proxy.rs
> @@ -17,6 +17,7 @@ use openssl::ssl::SslAcceptor;
>   use serde_json::{json, Value};
>   
>   use proxmox_http::Body;
> +use proxmox_http::RateLimiterTag;
>   use proxmox_lang::try_block;
>   use proxmox_router::{RpcEnvironment, RpcEnvironmentType};
>   use proxmox_sys::fs::CreateOptions;
> @@ -955,6 +956,7 @@ async fn run_traffic_control_updater() {
>   
>   fn lookup_rate_limiter(
>       peer: std::net::SocketAddr,
> +    tags: &[RateLimiterTag],
>   ) -> (Option<SharedRateLimit>, Option<SharedRateLimit>) {
>       let mut cache = TRAFFIC_CONTROL_CACHE.lock().unwrap();
>   
> @@ -962,7 +964,15 @@ fn lookup_rate_limiter(
>   
>       cache.reload(now);
>   
> -    let (_rule_name, read_limiter, write_limiter) = cache.lookup_rate_limiter(peer, now);
> +    let user = tags.iter().find_map(|tag| match tag {
> +        RateLimiterTag::User(user) => Some(user.as_str()),
> +    });
> +
> +    let authid = user.and_then(|s| s.parse::<pbs_api_types::Authid>().ok());
> +    let user_parsed = authid.as_ref().map(|auth_id| auth_id.user());
> +
> +    let (_rule_name, read_limiter, write_limiter) =
> +        cache.lookup_rate_limiter(peer, now, user_parsed);
>   
>       (read_limiter, write_limiter)
>   }
> diff --git a/src/traffic_control_cache.rs b/src/traffic_control_cache.rs
> index 830a8c04..95a7bf0b 100644
> --- a/src/traffic_control_cache.rs
> +++ b/src/traffic_control_cache.rs
> @@ -13,7 +13,7 @@ use proxmox_section_config::SectionConfigData;
>   
>   use proxmox_time::{parse_daily_duration, DailyDuration, TmEditor};
>   
> -use pbs_api_types::TrafficControlRule;
> +use pbs_api_types::{TrafficControlRule, Userid};
>   
>   use pbs_config::ConfigVersionCache;
>   
> @@ -31,6 +31,12 @@ struct ParsedTcRule {
>       timeframe: Vec<DailyDuration>, // parsed timeframe
>   }
>   
> +/// Represents a matched traffic control rule with its network prefix length and user rule status.
> +///
> +/// Used during rule lookup to track the best matching rule, where user-specific rules
> +/// take precedence over IP-only rules, and longer network prefixes are preferred.
> +type RuleMatch<'a> = (&'a ParsedTcRule, u8, bool); // (rule, network_prefix_length, is_user_rule)
> +
>   /// Traffic control statistics
>   pub struct TrafficStat {
>       /// Total incoming traffic (bytes)
> @@ -322,6 +328,7 @@ impl TrafficControlCache {
>       ///
>       /// - Rules where timeframe does not match are skipped.
>       /// - Rules with smaller network size have higher priority.
> +    /// - Rules with users are prioritized over IP-only rules, if multiple match
>       ///
>       /// Behavior is undefined if more than one rule matches after
>       /// above selection.
> @@ -329,10 +336,14 @@ impl TrafficControlCache {
>           &self,
>           peer: SocketAddr,
>           now: i64,
> +        user: Option<&Userid>,
>       ) -> (&str, Option<SharedRateLimit>, Option<SharedRateLimit>) {
>           let peer_ip = canonical_ip(peer.ip());
>   
> -        log::debug!("lookup_rate_limiter: {:?}", peer_ip);
> +        log::debug!(
> +            "lookup_rate_limiter: {peer_ip:?} - {}",
> +            user.map_or("(no user)", |u| u.as_str())
> +        );
>   
>           let now = match TmEditor::with_epoch(now, self.use_utc) {
>               Ok(now) => now,
> @@ -342,19 +353,33 @@ impl TrafficControlCache {
>               }
>           };
>   
> -        let mut last_rule_match = None;
> +        let mut last_rule_match: Option<RuleMatch> = None;
>   
>           for rule in self.rules.iter() {
>               if !timeframe_match(&rule.timeframe, &now) {
>                   continue;
>               }
>   
> +            if let Some(ref rule_users) = rule.config.users {
> +                if let Some(cur_user) = user {
> +                    if !rule_users.iter().any(|u| u == cur_user) {
> +                        continue;
> +                    }
> +                } else {
> +                    continue;
> +                }
> +            }
> +
>               if let Some(match_len) = network_match_len(&rule.networks, &peer_ip) {
> +                let is_user_rule = rule.config.users.is_some();
>                   match last_rule_match {
> -                    None => last_rule_match = Some((rule, match_len)),
> -                    Some((_, last_len)) => {
> -                        if match_len > last_len {
> -                            last_rule_match = Some((rule, match_len));
> +                    None => last_rule_match = Some((rule, match_len, is_user_rule)),
> +                    Some((_, last_len, last_is_user)) => {
> +                        // Prefer rules with users over IP-only rules; for same class use longest prefix
> +                        if (is_user_rule && !last_is_user)
> +                            || (is_user_rule == last_is_user && match_len > last_len)
> +                        {
> +                            last_rule_match = Some((rule, match_len, is_user_rule));
>                           }
>                       }
>                   }
> @@ -362,7 +387,7 @@ impl TrafficControlCache {
>           }
>   
>           match last_rule_match {
> -            Some((rule, _)) => {
> +            Some((rule, _, _)) => {
>                   match self.limiter_map.get(&rule.config.name) {
>                       Some((read_limiter, write_limiter)) => (
>                           &rule.config.name,
> @@ -454,34 +479,88 @@ rule: somewhere
>           let somewhere = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 1234);
>   
>           let (rule, read_limiter, write_limiter) =
> -            cache.lookup_rate_limiter(somewhere, THURSDAY_80_00);
> +            cache.lookup_rate_limiter(somewhere, THURSDAY_80_00, None);
>           assert_eq!(rule, "somewhere");
>           assert!(read_limiter.is_some());
>           assert!(write_limiter.is_some());
>   
> -        let (rule, read_limiter, write_limiter) = cache.lookup_rate_limiter(local, THURSDAY_19_00);
> +        let (rule, read_limiter, write_limiter) =
> +            cache.lookup_rate_limiter(local, THURSDAY_19_00, None);
>           assert_eq!(rule, "rule2");
>           assert!(read_limiter.is_some());
>           assert!(write_limiter.is_some());
>   
>           let (rule, read_limiter, write_limiter) =
> -            cache.lookup_rate_limiter(gateway, THURSDAY_15_00);
> +            cache.lookup_rate_limiter(gateway, THURSDAY_15_00, None);
>           assert_eq!(rule, "rule1");
>           assert!(read_limiter.is_some());
>           assert!(write_limiter.is_some());
>   
>           let (rule, read_limiter, write_limiter) =
> -            cache.lookup_rate_limiter(gateway, THURSDAY_19_00);
> +            cache.lookup_rate_limiter(gateway, THURSDAY_19_00, None);
>           assert_eq!(rule, "somewhere");
>           assert!(read_limiter.is_some());
>           assert!(write_limiter.is_some());
>   
>           let (rule, read_limiter, write_limiter) =
> -            cache.lookup_rate_limiter(private, THURSDAY_19_00);
> +            cache.lookup_rate_limiter(private, THURSDAY_19_00, None);
>           assert_eq!(rule, "rule2");
>           assert!(read_limiter.is_some());
>           assert!(write_limiter.is_some());
>   
>           Ok(())
>       }
> +
> +    #[test]
> +    fn test_user_based_rule_match_and_precedence() -> Result<(), Error> {
> +        // rule user1: user-specific for alice@pam on 192.168.2.0/24
> +        // rule ip1: ip-only broader network also matches, but user rule should win
> +        // rule user2: user-specific for bob@pam but different network shouldn't match
> +        let config_data = "
> +rule: user1
> +	comment user rule for alice
> +	network 192.168.2.0/24
> +	rate-in 50000000
> +	rate-out 50000000
> +	users alice@pam
> +
> +rule: ip1
> +	network 192.168.2.0/24
> +	rate-in 100000000
> +	rate-out 100000000
> +
> +rule: user2
> +	network 10.0.0.0/8
> +	rate-in 75000000
> +	rate-out 75000000
> +	users bob@pam
> +";
> +
> +        let config = pbs_config::traffic_control::CONFIG.parse("testconfig", config_data)?;
> +
> +        let mut cache = TrafficControlCache::new();
> +        cache.use_utc = true;
> +        cache.use_shared_memory = false; // avoid permission problems in test environment
> +
> +        cache.update_config(&config)?;
> +
> +        const NOW: i64 = make_test_time(0, 12, 0);
> +        let peer = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 2, 55)), 1234);
> +
> +        // No user -> should match ip1
> +        let (rule, _, _) = cache.lookup_rate_limiter(peer, NOW, None);
> +        assert_eq!(rule, "ip1");
> +
> +        // alice@pam -> should match user1 and take precedence over ip1
> +        let alice = Userid::try_from("alice@pam".to_string())?;
> +        let (rule, _, _) = cache.lookup_rate_limiter(peer, NOW, Some(&alice));
> +        assert_eq!(rule, "user1");
> +
> +        // bob@pam on same peer/network -> user2 is different network, should fall back to ip1
> +        let bob = Userid::try_from("bob@pam".to_string())?;
> +        let (rule, _, _) = cache.lookup_rate_limiter(peer, NOW, Some(&bob));
> +        assert_eq!(rule, "ip1");
> +
> +        Ok(())
> +    }
>   }



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox{, -backup} v3 0/6] add user specific rate-limits

[Christian Ebner]

On 11/10/25 2:43 PM, Hannes Laimer wrote:
> When a connection is accepted we create a shared tag handle for its
> rate-limited stream. The REST layer clears that handle before every
> request. Once a request authenticates successfully, we push a
> User(...) tag with the auth ID. Failed or unauthenticated requests
> leave the tag list empty. RateLimitedStream watches that handle and
> forces an immediate limiter refresh whenever the tag set changes so
> user-specific throttles take effect right away.
> 
> Currently rules with a user specified take priority over others. So:
> user > IP only > neither, in case two rules match.
> 
> If users and networks are specified, the rule only applies if both
> match. So, Any of the specified user connect from any of the specified
> network.
> 
> And all of this ofc still only if the given timeframe matches.
> 
> I did also test this with a basic nginx reverse proxy configured with
> `keepalive 32`, I didn't run into problems using this setup.
> 
> v3, thanks @Chris!:
>   - simplify code by passing the taglist to the callback, as sugested by
>     Chris
>   - mention potential future use-case in commit message
>   - created documented type for 3-tuple and inlined var for printing
> 
> v2, thanks @Chris!:
>   - fix problem with tag staying on connection after request finishes,
>     and with when it would be set in first place
>   - use a more generic tag-list on the connection, this is more general
>   - tag is now an enum, like chris suggested, this should make it
>     somewhat easy to extend if we at some point should want to
>


- tested per-user rate limits are set
- tested rate limits are applied for all users defined in the ruleset
- tested rate limits are honored with reverse proxy (haproxy with 
`http-reuse always`)
- checked correct per-user rules are applied when connections go trough 
proxy


With all the comments addressed, please consider:

Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@proxmox.com>


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] superseded: [PATCH proxmox{, -backup} v3 0/6] add user specific rate-limits

[Hannes Laimer]

superseded-by: 
https://lore.proxmox.com/pbs-devel/20251112103505.122844-1-h.laimer@proxmox.com/T/#t

On 11/10/25 14:43, Hannes Laimer wrote:
> When a connection is accepted we create a shared tag handle for its
> rate-limited stream. The REST layer clears that handle before every
> request. Once a request authenticates successfully, we push a
> User(...) tag with the auth ID. Failed or unauthenticated requests
> leave the tag list empty. RateLimitedStream watches that handle and
> forces an immediate limiter refresh whenever the tag set changes so
> user-specific throttles take effect right away.
> 
> Currently rules with a user specified take priority over others. So:
> user > IP only > neither, in case two rules match.
> 
> If users and networks are specified, the rule only applies if both
> match. So, Any of the specified user connect from any of the specified
> network.
> 
> And all of this ofc still only if the given timeframe matches.
> 
> I did also test this with a basic nginx reverse proxy configured with
> `keepalive 32`, I didn't run into problems using this setup.
> 
> v3, thanks @Chris!:
>   - simplify code by passing the taglist to the callback, as sugested by
>     Chris
>   - mention potential future use-case in commit message
>   - created documented type for 3-tuple and inlined var for printing
> 
> v2, thanks @Chris!:
>   - fix problem with tag staying on connection after request finishes,
>     and with when it would be set in first place
>   - use a more generic tag-list on the connection, this is more general
>   - tag is now an enum, like chris suggested, this should make it
>     somewhat easy to extend if we at some point should want to
> 
> 
> proxmox:
> 
> Hannes Laimer (3):
>    pbs-api-types: allow traffic-control rules to match users
>    http: track user tag updates on rate-limited streams
>    rest-server: propagate rate-limit tags from authenticated users
> 
>   pbs-api-types/src/traffic_control.rs    |   9 ++
>   proxmox-http/src/lib.rs                 |   2 +-
>   proxmox-http/src/rate_limited_stream.rs |  40 ++++++-
>   proxmox-rest-server/src/connection.rs   |  11 +-
>   proxmox-rest-server/src/rest.rs         | 137 +++++++++++++++++++++++-
>   5 files changed, 186 insertions(+), 13 deletions(-)
> 
> 
> proxmox-backup:
> 
> Hannes Laimer (3):
>    api: taffic-control: update/delete users on rule correctly
>    traffic-control: handle users specified in a rule correctly
>    ui: traffic-control: add users field in edit form and list
> 
>   src/api2/config/traffic_control.rs |   8 +++
>   src/bin/proxmox-backup-proxy.rs    |  12 +++-
>   src/traffic_control_cache.rs       | 105 +++++++++++++++++++++++++----
>   www/config/TrafficControlView.js   |   7 ++
>   www/window/TrafficControlEdit.js   |  18 +++++
>   5 files changed, 136 insertions(+), 14 deletions(-)
> 
> 
> Summary over all repositories:
>    10 files changed, 322 insertions(+), 27 deletions(-)
> 



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel