fix: Block unlimited users for admins with max limits

Author: ImMohammad20000

app/operation/user.py

@@ -281,6 +281,7 @@ async def _persist_bulk_users(
                     admin,
                     data_limit=user_to_create.data_limit,
                     expire=user_to_create.expire,
+                    status=user_to_create.status,
                     on_hold_expire_duration=user_to_create.on_hold_expire_duration,
                     hwid_limit=user_to_create.hwid_limit,
                     data_limit_reset_strategy=user_to_create.data_limit_reset_strategy,
@@ -374,11 +375,15 @@ async def _enforce_user_limits(
         *,
         data_limit: int | None = None,
         expire: dt | int | None = None,
+        status: UserStatus | None = None,
         on_hold_expire_duration: int | None = None,
         hwid_limit: int | None = None,
         data_limit_reset_strategy=None,
         next_plan=None,
         check_max_users: bool = False,
+        require_finite_data_limit: bool = True,
+        require_finite_expire: bool = True,
+        require_finite_hwid_limit: bool = True,
     ) -> None:
         """Enforce role-level limits and feature flags. No-op for owner."""
         if admin.is_owner:
@@ -391,6 +396,13 @@ async def _enforce_user_limits(
             if current_count >= limits.max_users:
                 await self.raise_error(message=f"User limit reached ({limits.max_users})", code=400, db=db)
 
+        if limits.data_limit_max is not None and require_finite_data_limit and (data_limit is None or data_limit <= 0):
+            await self.raise_error(
+                message=f"Data limit cannot be unlimited; maximum is {readable_size(limits.data_limit_max)}",
+                code=400,
+                db=db,
+            )
+
         if data_limit is not None and data_limit > 0:
             if limits.data_limit_min is not None and data_limit < limits.data_limit_min:
                 await self.raise_error(
@@ -401,6 +413,26 @@ async def _enforce_user_limits(
                     message=f"Data limit cannot exceed {readable_size(limits.data_limit_max)}", code=400, db=db
                 )
 
+        status_value = getattr(status, "value", status)
+        uses_on_hold_duration = status_value == UserStatus.on_hold.value
+
+        if limits.expire_max is not None and require_finite_expire:
+            if uses_on_hold_duration:
+                if on_hold_expire_duration is None or on_hold_expire_duration <= 0:
+                    await self.raise_error(
+                        message=(
+                            f"On-hold duration cannot be unlimited; maximum is {readable_duration(limits.expire_max)}"
+                        ),
+                        code=400,
+                        db=db,
+                    )
+            elif expire is None or expire == 0:
+                await self.raise_error(
+                    message=f"Expire cannot be unlimited; maximum is {readable_duration(limits.expire_max)} from now",
+                    code=400,
+                    db=db,
+                )
+
         if expire is not None and expire != 0:
             expire_dt = fix_datetime_timezone(expire)
             seconds = (expire_dt - datetime.now(timezone.utc)).total_seconds()
@@ -431,6 +463,17 @@ async def _enforce_user_limits(
                     db=db,
                 )
 
+        if (
+            limits.max_hwid_per_user is not None
+            and require_finite_hwid_limit
+            and (hwid_limit is None or hwid_limit <= 0)
+        ):
+            await self.raise_error(
+                message=f"HWID limit cannot be unlimited; maximum is {limits.max_hwid_per_user}",
+                code=400,
+                db=db,
+            )
+
         if hwid_limit is not None:
             if limits.min_hwid_per_user is not None and hwid_limit < limits.min_hwid_per_user:
                 await self.raise_error(
@@ -456,25 +499,26 @@ async def create_user(
         if new_user.hwid_limit is None:
             new_user.hwid_limit = hwid_conf.fallback_limit
 
-        if new_user.hwid_limit is not None and not admin.is_owner:
-            if new_user.hwid_limit < hwid_conf.min_limit:
-                await self.raise_error(message=f"HWID limit cannot be less than {hwid_conf.min_limit}", code=400, db=db)
-            if hwid_conf.max_limit > 0 and (new_user.hwid_limit > hwid_conf.max_limit or new_user.hwid_limit == 0):
-                await self.raise_error(message=f"HWID limit cannot exceed {hwid_conf.max_limit}", code=400, db=db)
-
         if not skip_role_limits:
             await self._enforce_user_limits(
                 db,
                 admin,
                 data_limit=new_user.data_limit,
                 expire=new_user.expire,
+                status=new_user.status,
                 on_hold_expire_duration=new_user.on_hold_expire_duration,
                 hwid_limit=new_user.hwid_limit,
                 data_limit_reset_strategy=new_user.data_limit_reset_strategy,
                 next_plan=new_user.next_plan,
                 check_max_users=True,
             )
 
+        if new_user.hwid_limit is not None and not admin.is_owner:
+            if new_user.hwid_limit < hwid_conf.min_limit:
+                await self.raise_error(message=f"HWID limit cannot be less than {hwid_conf.min_limit}", code=400, db=db)
+            if hwid_conf.max_limit > 0 and (new_user.hwid_limit > hwid_conf.max_limit or new_user.hwid_limit == 0):
+                await self.raise_error(message=f"HWID limit cannot exceed {hwid_conf.max_limit}", code=400, db=db)
+
         if new_user.next_plan is not None and new_user.next_plan.user_template_id is not None:
             await self.get_validated_user_template(db, new_user.next_plan.user_template_id)
 
@@ -513,27 +557,61 @@ async def _prepare_modified_user(
                     db=db,
                 )
 
-        if modified_user.hwid_limit is not None and not admin.is_owner:
-            hwid_conf = await hwid_settings()
-            if modified_user.hwid_limit < hwid_conf.min_limit:
-                await self.raise_error(message=f"HWID limit cannot be less than {hwid_conf.min_limit}", code=400, db=db)
-            if hwid_conf.max_limit > 0 and (
-                modified_user.hwid_limit > hwid_conf.max_limit or modified_user.hwid_limit == 0
-            ):
-                await self.raise_error(message=f"HWID limit cannot exceed {hwid_conf.max_limit}", code=400, db=db)
-
         if not skip_role_limits:
+            modified_fields = modified_user.model_fields_set
+            data_limit_was_changed = "data_limit" in modified_fields and modified_user.data_limit is not None
+            expire_was_changed = "expire" in modified_fields and modified_user.expire is not None
+            status_was_changed = modified_user.status is not None
+            modified_status_value = getattr(modified_user.status, "value", modified_user.status)
+            status_requires_finite_expire = modified_status_value in {
+                UserStatus.active.value,
+                UserStatus.on_hold.value,
+            }
+            on_hold_duration_was_changed = (
+                "on_hold_expire_duration" in modified_fields and modified_user.on_hold_expire_duration is not None
+            )
+            expire_requires_finite_limit = (
+                expire_was_changed
+                or (status_was_changed and status_requires_finite_expire)
+                or on_hold_duration_was_changed
+            )
+            hwid_limit_was_changed = "hwid_limit" in modified_fields and modified_user.hwid_limit is not None
+
+            effective_status = modified_user.status if modified_user.status is not None else db_user.status
+            effective_expire = modified_user.expire
+            effective_on_hold_expire_duration = modified_user.on_hold_expire_duration
+            if status_was_changed and status_requires_finite_expire:
+                if modified_status_value == UserStatus.on_hold.value:
+                    effective_expire = None
+                    if effective_on_hold_expire_duration is None:
+                        effective_on_hold_expire_duration = db_user.on_hold_expire_duration
+                elif effective_expire is None:
+                    effective_expire = db_user.expire
+
             await self._enforce_user_limits(
                 db,
                 admin,
                 data_limit=modified_user.data_limit,
-                expire=modified_user.expire,
-                on_hold_expire_duration=modified_user.on_hold_expire_duration,
+                expire=effective_expire,
+                status=effective_status,
+                on_hold_expire_duration=effective_on_hold_expire_duration,
                 hwid_limit=modified_user.hwid_limit,
                 data_limit_reset_strategy=modified_user.data_limit_reset_strategy,
                 next_plan=modified_user.next_plan,
+                require_finite_data_limit=data_limit_was_changed,
+                require_finite_expire=expire_requires_finite_limit,
+                require_finite_hwid_limit=hwid_limit_was_changed,
             )
 
+        if modified_user.hwid_limit is not None and not admin.is_owner:
+            hwid_conf = await hwid_settings()
+            if modified_user.hwid_limit < hwid_conf.min_limit:
+                await self.raise_error(message=f"HWID limit cannot be less than {hwid_conf.min_limit}", code=400, db=db)
+            if hwid_conf.max_limit > 0 and (
+                modified_user.hwid_limit > hwid_conf.max_limit or modified_user.hwid_limit == 0
+            ):
+                await self.raise_error(message=f"HWID limit cannot exceed {hwid_conf.max_limit}", code=400, db=db)
+
         validated_groups = None
         if modified_user.group_ids:
             validated_groups = await self.validate_all_groups(db, modified_user)
@@ -1250,7 +1328,14 @@ async def create_user_from_template(
             raise exc
 
         # Template defines data_limit/expire/etc — only check max_users
-        await self._enforce_user_limits(db, admin, check_max_users=True)
+        await self._enforce_user_limits(
+            db,
+            admin,
+            check_max_users=True,
+            require_finite_data_limit=False,
+            require_finite_expire=False,
+            require_finite_hwid_limit=False,
+        )
         return await self.create_user(db, new_user, admin, skip_role_limits=True)
 
     async def _modify_user_with_template(

tests/api/test_user.py

@@ -99,6 +99,34 @@ def _build_v2_subscription_token(user_id: int, secret: str) -> str:
     return data_b64 + sign
 
 
+def _create_limited_user_creator_role(access_token: str) -> dict:
+    response = client.post(
+        "/api/admin-role",
+        headers=auth_headers(access_token),
+        json={
+            "name": unique_name("bounded_user_creator"),
+            "permissions": {"users": {"create": True, "read": True, "update": True, "delete": True}},
+            "limits": {
+                "max_users": None,
+                "data_limit_min": None,
+                "data_limit_max": 10 * 1024 * 1024,
+                "expire_min": None,
+                "expire_max": 24 * 60 * 60,
+                "min_hwid_per_user": None,
+                "max_hwid_per_user": 3,
+            },
+            "features": {"can_use_reset_strategy": True, "can_use_next_plan": True},
+            "access": {"require_template": False, "allowed_template_ids": None, "allowed_group_ids": None},
+        },
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    return response.json()
+
+
+def _delete_role(access_token: str, role_id: int) -> None:
+    client.delete(f"/api/admin-role/{role_id}", headers=auth_headers(access_token))
+
+
 def test_subscription_token_generation_avoids_trailing_dash_or_underscore_and_keeps_v2_compatibility(monkeypatch):
     secret = "test-secret"
 
@@ -152,6 +180,94 @@ def test_user_create_active(access_token):
         cleanup_groups(access_token, core, groups)
 
 
+def test_limited_admin_cannot_create_or_modify_user_to_unlimited_data_or_expire(access_token):
+    role = _create_limited_user_creator_role(access_token)
+    admin = create_admin(access_token, role_id=role["id"])
+    admin_token = _login(admin["username"], admin["password"])
+    username = unique_name("bounded_limit_user")
+    finite_expire = (datetime.now(timezone.utc).replace(microsecond=0) + timedelta(hours=1)).isoformat()
+
+    try:
+        response = client.post(
+            "/api/user",
+            headers=auth_headers(admin_token),
+            json={
+                "username": unique_name("bounded_no_data"),
+                "proxy_settings": {},
+                "expire": finite_expire,
+                "data_limit": 0,
+                "data_limit_reset_strategy": "no_reset",
+                "status": "active",
+            },
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "Data limit cannot be unlimited" in response.json()["detail"]
+
+        response = client.post(
+            "/api/user",
+            headers=auth_headers(admin_token),
+            json={
+                "username": unique_name("bounded_no_expire"),
+                "proxy_settings": {},
+                "expire": 0,
+                "data_limit": 1024 * 1024,
+                "data_limit_reset_strategy": "no_reset",
+                "status": "active",
+            },
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "Expire cannot be unlimited" in response.json()["detail"]
+
+        response = client.post(
+            "/api/user",
+            headers=auth_headers(admin_token),
+            json={
+                "username": unique_name("bounded_no_hwid"),
+                "proxy_settings": {},
+                "expire": finite_expire,
+                "data_limit": 1024 * 1024,
+                "data_limit_reset_strategy": "no_reset",
+                "hwid_limit": 0,
+                "status": "active",
+            },
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "HWID limit cannot be unlimited" in response.json()["detail"]
+
+        response = client.post(
+            "/api/user",
+            headers=auth_headers(admin_token),
+            json={
+                "username": username,
+                "proxy_settings": {},
+                "expire": finite_expire,
+                "data_limit": 1024 * 1024,
+                "data_limit_reset_strategy": "no_reset",
+                "status": "active",
+            },
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+
+        response = client.put(f"/api/user/{username}", headers=auth_headers(admin_token), json={"note": "allowed"})
+        assert response.status_code == status.HTTP_200_OK
+
+        response = client.put(f"/api/user/{username}", headers=auth_headers(admin_token), json={"data_limit": 0})
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "Data limit cannot be unlimited" in response.json()["detail"]
+
+        response = client.put(f"/api/user/{username}", headers=auth_headers(admin_token), json={"expire": 0})
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "Expire cannot be unlimited" in response.json()["detail"]
+
+        response = client.put(f"/api/user/{username}", headers=auth_headers(admin_token), json={"hwid_limit": 0})
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "HWID limit cannot be unlimited" in response.json()["detail"]
+    finally:
+        client.delete(f"/api/user/{username}", headers=auth_headers(access_token))
+        delete_admin(access_token, admin["username"])
+        _delete_role(access_token, role["id"])
+
+
 def test_user_create_expire_timezone_offset_normalized_to_utc(access_token):
     """Expire with non-UTC offset should be persisted as the same UTC instant."""
     core, groups = setup_groups(access_token, 1)