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

[pbs-devel] [PATCH proxmox{, -backup} v4 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.

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


v4, thanks @Chris:
 - format
 - added and reordered trailers
 - improve commit message
 - no code changes from v3

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 |  46 +++++++-
 proxmox-rest-server/src/connection.rs   |  11 +-
 proxmox-rest-server/src/rest.rs         | 137 +++++++++++++++++++++++-
 5 files changed, 192 insertions(+), 13 deletions(-)


proxmox-backup:

Hannes Laimer (3):
  api: taffic-control: update/delete users on rule correctly
  traffic-control: add user-specific rule matching and precedence
  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, 328 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 v4 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.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@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..174f53a2 100644
--- a/pbs-api-types/src/traffic_control.rs
+++ b/pbs-api-types/src/traffic_control.rs
@@ -5,6 +5,7 @@ use proxmox_schema::{api, ApiType, Schema, StringSchema, Updater};
 
 use proxmox_schema::api_types::CIDR_SCHEMA;
 
+use crate::Userid;
 use crate::{DAILY_DURATION_FORMAT, PROXMOX_SAFE_ID_FORMAT, SINGLE_LINE_COMMENT_SCHEMA};
 
 pub const TRAFFIC_CONTROL_TIMEFRAME_SCHEMA: Schema =
@@ -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 v4 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>
Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@proxmox.com>
---
 proxmox-http/src/lib.rs                 |  2 +-
 proxmox-http/src/rate_limited_stream.rs | 46 ++++++++++++++++++++++---
 2 files changed, 42 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..f4d02eb0 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,15 @@ 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 +91,33 @@ 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 +131,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 v4 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>
Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@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..da55ea25 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 v4 1/3] api: taffic-control: update/delete users on rule correctly

[Hannes Laimer]

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@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 v4 2/3] traffic-control: add user-specific rule matching and precedence

[Hannes Laimer]

Introduce user-aware rule lookup by accepting an optional user and preferring
rules that specify 'users' over IP-only matches. For rules of the same class,
keep longest-prefix selection.

Pass the authenticated user from the proxy via RateLimiterTag::User to
lookup_rate_limiter(), and preserve behavior when no user is provided.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@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 v4 3/3] ui: traffic-control: add users field in edit form and list

[Hannes Laimer]

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
Reviewed-by: Christian Ebner <c.ebner@proxmox.com>
Tested-by: Christian Ebner <c.ebner@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..6079ed6c 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