Add user limits.

MirahImage · MirahImage · commit a63c258393f9 · 2025-03-19T17:06:58.000+01:00
diff --git a/api/v1beta1/user_types.go b/api/v1beta1/user_types.go
@@ -37,6 +37,11 @@ type UserSpec struct {
 	//
 	// Note that this import only occurs at creation time, and is ignored once a password has been set on a User.
 	ImportCredentialsSecret *corev1.LocalObjectReference `json:"importCredentialsSecret,omitempty"`
+	// Limits to apply to a user to restrict the number of connections and channels
+	// the user can create. These limits can be used as guard rails in environments
+	// where applications cannot be trusted and monitored in detail, for example,
+	// when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
+	UserLimits UserLimits `json:"limits,omitempty"`
 }
 
 // UserStatus defines the observed state of User.
@@ -56,6 +61,17 @@ type UserStatus struct {
 // +kubebuilder:validation:Enum=management;policymaker;monitoring;administrator
 type UserTag string
 
+// Limits to apply to a user to restrict the number of connections and channels
+// the user can create. These limits can be used as guard rails in environments
+// where applications cannot be trusted and monitored in detail, for example,
+// when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
+type UserLimits struct {
+	// Limits how many connections the user can open.
+	Connections int32 `json:"connections,omitempty"`
+	// Limits how many AMQP 0.9.1 channels the user can open.
+	Channels int32 `json:"channels,omitempty"`
+}
+
 // +genclient
 // +kubebuilder:object:root=true
 // +kubebuilder:resource:categories=rabbitmq
diff --git a/api/v1beta1/user_types_test.go b/api/v1beta1/user_types_test.go
@@ -92,4 +92,46 @@ var _ = Describe("user spec", func() {
 		})
 	})
 
+	When("creating a user with limits", func() {
+		var user User
+		var username string
+		var userLimits UserLimits
+
+		JustBeforeEach(func() {
+			user = User{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      username,
+					Namespace: namespace,
+				},
+				Spec: UserSpec{
+					RabbitmqClusterReference: RabbitmqClusterReference{
+						Name: "some-cluster",
+					},
+					UserLimits: userLimits,
+				},
+			}
+		})
+
+		When("creating a user with valid limits", func() {
+			BeforeEach(func() {
+				username = "limits-user"
+				userLimits = UserLimits{
+					Connections: 5,
+					Channels:    10,
+				}
+			})
+			It("successfully creates the user", func() {
+				Expect(k8sClient.Create(ctx, &user)).To(Succeed())
+				fetchedUser := &User{}
+				Expect(k8sClient.Get(ctx, types.NamespacedName{
+					Name:      user.Name,
+					Namespace: user.Namespace,
+				}, fetchedUser)).To(Succeed())
+				Expect(fetchedUser.Spec.RabbitmqClusterReference).To(Equal(RabbitmqClusterReference{
+					Name: "some-cluster",
+				}))
+				Expect(fetchedUser.Spec.UserLimits).To(Equal(UserLimits{Connections: 5, Channels: 10}))
+			})
+		})
+	})
 })
diff --git a/api/v1beta1/zz_generated.deepcopy.go b/api/v1beta1/zz_generated.deepcopy.go
diff --git a/config/crd/bases/rabbitmq.com_users.yaml b/config/crd/bases/rabbitmq.com_users.yaml
@@ -64,6 +64,23 @@ spec:
                     type: string
                 type: object
                 x-kubernetes-map-type: atomic
+              limits:
+                description: |-
+                  Limits to apply to a user to restrict the number of connections and channels
+                  the user can create. These limits can be used as guard rails in environments
+                  where applications cannot be trusted and monitored in detail, for example,
+                  when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
+                properties:
+                  channels:
+                    description: Limits how many AMQP 0.9.1 channels the user can
+                      open.
+                    format: int32
+                    type: integer
+                  connections:
+                    description: Limits how many connections the user can open.
+                    format: int32
+                    type: integer
+                type: object
               rabbitmqClusterReference:
                 description: |-
                   Reference to the RabbitmqCluster that the user will be created for. This cluster must
diff --git a/controllers/permission_controller_test.go b/controllers/permission_controller_test.go
@@ -261,6 +261,14 @@ var _ = Describe("permission-controller", func() {
 				Status:     "204 No Content",
 				StatusCode: http.StatusNoContent,
 			}, nil)
+			fakeRabbitMQClient.PutUserLimitsReturns(&http.Response{
+				Status:     "201 Created",
+				StatusCode: http.StatusCreated,
+			}, nil)
+			fakeRabbitMQClient.DeleteUserLimitsReturns(&http.Response{
+				Status:     "204 No Content",
+				StatusCode: http.StatusNoContent,
+			}, nil)
 		})
 
 		Context("creation", func() {
diff --git a/controllers/user_controller.go b/controllers/user_controller.go
@@ -221,7 +221,14 @@ func (r *UserReconciler) DeclareFunc(ctx context.Context, client rabbitmqclient.
 	}
 	logger.Info("Generated user settings", "user", user.Name, "settings", userSettings)
 
-	return validateResponse(client.PutUser(userSettings.Name, userSettings))
+	err = validateResponse(client.PutUser(userSettings.Name, userSettings))
+	if err != nil {
+		return err
+	}
+
+	userLimits := internal.GenerateUserLimits(user.Spec.UserLimits)
+	logger.Info("Generated user limits", "user", user.Name, "limits", userLimits)
+	return validateResponse(client.PutUserLimits(user.Name, userLimits))
 }
 
 func (r *UserReconciler) getUserCredentials(ctx context.Context, user *topology.User) (*corev1.Secret, error) {
@@ -245,5 +252,13 @@ func (r *UserReconciler) DeleteFunc(ctx context.Context, client rabbitmqclient.C
 	} else if err != nil {
 		return err
 	}
+
+	userLimits := []string{"max-connections", "max-channels"}
+	err = validateResponseForDeletion(client.DeleteUserLimits(user.Status.Username, userLimits))
+	if errors.Is(err, NotFound) {
+		logger.Info("cannot find user limits in rabbitmq server; already deleted", "user", user.Name, "limits", userLimits)
+	} else if err != nil {
+		return err
+	}
 	return nil
 }
diff --git a/controllers/user_controller_test.go b/controllers/user_controller_test.go
@@ -33,6 +33,7 @@ var _ = Describe("UserController", func() {
 		managerCtx    context.Context
 		managerCancel context.CancelFunc
 		k8sClient     runtimeClient.Client
+		userLimits    topology.UserLimits
 	)
 
 	BeforeEach(func() {
@@ -91,6 +92,7 @@ var _ = Describe("UserController", func() {
 				RabbitmqClusterReference: topology.RabbitmqClusterReference{
 					Name: "example-rabbit",
 				},
+				UserLimits: userLimits,
 			},
 		}
 	})
@@ -154,6 +156,54 @@ var _ = Describe("UserController", func() {
 					})))
 			})
 		})
+
+		When("the user has limits defined", func() {
+			BeforeEach(func() {
+				userName = "test-user-limits"
+				userLimits = topology.UserLimits{
+					Connections: 5,
+					Channels:    10,
+				}
+				fakeRabbitMQClient.PutUserReturns(&http.Response{
+					Status:     "201 Created",
+					StatusCode: http.StatusCreated,
+				}, nil)
+				fakeRabbitMQClient.PutUserLimitsReturns(&http.Response{
+					Status:     "201 Created",
+					StatusCode: http.StatusCreated,
+				}, nil)
+			})
+
+			It("should create the user limits", func() {
+				Expect(k8sClient.Create(ctx, &user)).To(Succeed())
+				Eventually(func() []topology.Condition {
+					_ = k8sClient.Get(
+						ctx,
+						types.NamespacedName{Name: user.Name, Namespace: user.Namespace},
+						&user,
+					)
+
+					return user.Status.Conditions
+				}).
+					Within(statusEventsUpdateTimeout).
+					WithPolling(time.Second).
+					Should(ContainElement(MatchFields(IgnoreExtras, Fields{
+						"Type":   Equal(topology.ConditionType("Ready")),
+						"Reason": Equal("SuccessfulCreateOrUpdate"),
+						"Status": Equal(corev1.ConditionTrue),
+					})))
+				By("calling PutUserLimits with the correct user limits")
+				Expect(fakeRabbitMQClient.PutUserLimitsCallCount()).To(BeNumerically(">", 0))
+				username, userLimitsValues := fakeRabbitMQClient.PutUserLimitsArgsForCall(0)
+				Expect(username).To(Equal(userName))
+				connectionLimit, ok := userLimitsValues["max-connections"]
+				Expect(ok).To(BeTrue())
+				Expect(connectionLimit).To(Equal(5))
+				channelLimit, ok := userLimitsValues["max-channels"]
+				Expect(ok).To(BeTrue())
+				Expect(channelLimit).To(Equal(10))
+			})
+		})
 	})
 
 	When("deleting a user", func() {
@@ -162,6 +212,14 @@ var _ = Describe("UserController", func() {
 				Status:     "201 Created",
 				StatusCode: http.StatusCreated,
 			}, nil)
+			fakeRabbitMQClient.PutUserLimitsReturns(&http.Response{
+				Status:     "201 Created",
+				StatusCode: http.StatusCreated,
+			}, nil)
+			fakeRabbitMQClient.DeleteUserLimitsReturns(&http.Response{
+				Status:     "204 No Content",
+				StatusCode: http.StatusNoContent,
+			}, nil)
 			Expect(k8sClient.Create(ctx, &user)).To(Succeed())
 			Eventually(func() []topology.Condition {
 				_ = k8sClient.Get(
diff --git a/docs/api/rabbitmq.com.ref.asciidoc b/docs/api/rabbitmq.com.ref.asciidoc
@@ -1366,6 +1366,27 @@ More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-
 |===
 
 
+[id="{anchor_prefix}-github-com-rabbitmq-messaging-topology-operator-api-v1beta1-userlimits"]
+==== UserLimits 
+
+Limits to apply to a user to restrict the number of connections and channels
+the user can create. These limits can be used as guard rails in environments
+where applications cannot be trusted and monitored in detail, for example,
+when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
+
+.Appears In:
+****
+- xref:{anchor_prefix}-github-com-rabbitmq-messaging-topology-operator-api-v1beta1-userspec[$$UserSpec$$]
+****
+
+[cols="25a,75a", options="header"]
+|===
+| Field | Description
+| *`connections`* __integer__ | Limits how many connections the user can open.
+| *`channels`* __integer__ | Limits how many AMQP 0.9.1 channels the user can open.
+|===
+
+
 [id="{anchor_prefix}-github-com-rabbitmq-messaging-topology-operator-api-v1beta1-userlist"]
 ==== UserList 
 
@@ -1422,6 +1443,10 @@ password will be generated. The Secret must have the following keys in its Data
  * `password` – Plain-text password. Will be used only if the `passwordHash` key is missing.
 
 Note that this import only occurs at creation time, and is ignored once a password has been set on a User.
+| *`limits`* __xref:{anchor_prefix}-github-com-rabbitmq-messaging-topology-operator-api-v1beta1-userlimits[$$UserLimits$$]__ | Limits to apply to a user to restrict the number of connections and channels
+the user can create. These limits can be used as guard rails in environments
+where applications cannot be trusted and monitored in detail, for example,
+when RabbitMQ clusters are offered as a service. See https://www.rabbitmq.com/docs/user-limits.
 |===
 
 
diff --git a/internal/user.go b/internal/user.go
@@ -62,3 +62,14 @@ func GenerateUserSettings(credentials *corev1.Secret, tags []topology.UserTag) (
 		HashingAlgorithm: rabbithole.HashingAlgorithmSHA512,
 	}, nil
 }
+
+func GenerateUserLimits(userLimits topology.UserLimits) rabbithole.UserLimitsValues {
+	userLimitsValues := rabbithole.UserLimitsValues{}
+	if userLimits.Connections > 0 {
+		userLimitsValues["max-connections"] = int(userLimits.Connections)
+	}
+	if userLimits.Channels > 0 {
+		userLimitsValues["max-channels"] = int(userLimits.Channels)
+	}
+	return userLimitsValues
+}
diff --git a/internal/user_test.go b/internal/user_test.go
@@ -63,4 +63,31 @@ var _ = Describe("GenerateUserSettings", func() {
 		// Password should not be sent, even if provided
 		Expect(settings.Password).To(BeEmpty())
 	})
+
+	When("user limits are provided", func() {
+		It("uses the limits to generate the expected rabbithole.UserLimits", func() {
+			limits := internal.GenerateUserLimits(topology.UserLimits{
+				Connections: 3,
+				Channels:    7,
+			})
+			Expect(limits["max-connections"]).To(Equal(3))
+			Expect(limits["max-channels"]).To(Equal(7))
+		})
+
+		It("does not create unspecified limits", func() {
+			limits := internal.GenerateUserLimits(topology.UserLimits{
+				Connections: 5,
+			})
+			Expect(limits["max-connections"]).To(Equal(5))
+			_, ok := limits["max-channels"]
+			Expect(ok).To(BeFalse())
+		})
+	})
+
+	When("no user limits are provided", func() {
+		It("does not specify limits", func() {
+			limits := internal.GenerateUserLimits(topology.UserLimits{})
+			Expect(limits).To(BeEmpty())
+		})
+	})
 })
diff --git a/rabbitmqclient/rabbitmq_client_factory.go b/rabbitmqclient/rabbitmq_client_factory.go
@@ -22,6 +22,8 @@ import (
 type Client interface {
 	PutUser(string, rabbithole.UserSettings) (*http.Response, error)
 	DeleteUser(string) (*http.Response, error)
+	PutUserLimits(string, rabbithole.UserLimitsValues) (*http.Response, error)
+	DeleteUserLimits(string, rabbithole.UserLimits) (*http.Response, error)
 	DeclareBinding(string, rabbithole.BindingInfo) (*http.Response, error)
 	DeleteBinding(string, rabbithole.BindingInfo) (*http.Response, error)
 	ListQueueBindingsBetween(string, string, string) ([]rabbithole.BindingInfo, error)
diff --git a/rabbitmqclient/rabbitmqclientfakes/fake_client.go b/rabbitmqclient/rabbitmqclientfakes/fake_client.go
diff --git a/system_tests/user_system_test.go b/system_tests/user_system_test.go
