feat(admin-role): add HWID policy to admin roles

.gitignore

@@ -171,3 +171,4 @@ deadlock-analysis.md
 
 # AI
 .AGENT
+graphify-out

app/db/crud/admin_role.py

@@ -52,6 +52,7 @@ async def create_role(db: AsyncSession, data: AdminRoleCreate) -> AdminRole:
         limits=data.limits.model_dump(),
         features=data.features.model_dump(),
         access=data.access.model_dump(),
+        hwid=data.hwid.model_dump(),
         disabled_when_limited=data.disabled_when_limited,
         disable_users_when_limited=data.disable_users_when_limited,
     )
@@ -74,6 +75,8 @@ async def modify_role(db: AsyncSession, role: AdminRole, data: AdminRoleModify)
         role.features = data.features.model_dump()
     if data.access is not None:
         role.access = data.access.model_dump()
+    if data.hwid is not None:
+        role.hwid = data.hwid.model_dump()
     if data.disabled_when_limited is not None:
         role.disabled_when_limited = data.disabled_when_limited
     if data.disable_users_when_limited is not None:

app/db/migrations/env.py

@@ -1,7 +1,9 @@
 import asyncio
 from logging.config import fileConfig
+from sqlalchemy import JSON
 from sqlalchemy import BigInteger
 from sqlalchemy import pool
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.engine import Connection
 from sqlalchemy.ext.asyncio import async_engine_from_config
 from alembic import context
@@ -40,15 +42,26 @@ def _compare_type(context, inspected_column, metadata_column, inspected_type, me
     BIGINT depending on how the table was originally created, which can produce
     false-positive autogenerate diffs.
     """
-    if context.dialect.name != "sqlite":
-        return None
-
-    sqlite_bigint_equivalent = (
-        (isinstance(inspected_type, BigInteger) and isinstance(metadata_type, SqliteCompatibleBigInteger))
-        or (isinstance(inspected_type, SqliteCompatibleBigInteger) and isinstance(metadata_type, BigInteger))
-    )
-    if sqlite_bigint_equivalent:
-        return False
+    if context.dialect.name == "sqlite":
+        sqlite_bigint_equivalent = (
+            (isinstance(inspected_type, BigInteger) and isinstance(metadata_type, SqliteCompatibleBigInteger))
+            or (isinstance(inspected_type, SqliteCompatibleBigInteger) and isinstance(metadata_type, BigInteger))
+        )
+        if sqlite_bigint_equivalent:
+            return False
+
+    # PostgreSQL reflection can report JSON with explicit astext_type while
+    # metadata often renders as bare JSON(), which is not a schema change.
+    # Keep JSON vs JSONB detection intact by only bypassing plain JSON pairs.
+    if context.dialect.name == "postgresql":
+        is_plain_json_pair = (
+            isinstance(inspected_type, JSON)
+            and isinstance(metadata_type, JSON)
+            and not isinstance(inspected_type, JSONB)
+            and not isinstance(metadata_type, JSONB)
+        )
+        if is_plain_json_pair:
+            return False
 
     return None
 

app/db/migrations/versions/2c6e9d34a1f0_add_hwid_policy_to_admin_roles.py

@@ -0,0 +1,54 @@
+"""add hwid policy to admin roles
+
+Revision ID: 2c6e9d34a1f0
+Revises: bb4a32b7f5ce
+Create Date: 2026-05-22 00:00:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "2c6e9d34a1f0"
+down_revision = "bb4a32b7f5ce"
+branch_labels = None
+depends_on = None
+
+def upgrade() -> None:
+    with op.batch_alter_table("admin_roles", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "hwid",
+                sa.JSON(),
+                nullable=True,
+            )
+        )
+
+    # Cross-dialect safe JSON initialization (SQLite/MySQL/PostgreSQL)
+    admin_roles = sa.table(
+        "admin_roles",
+        sa.column("hwid", sa.JSON()),
+    )
+    op.execute(
+        admin_roles.update()
+        .where(admin_roles.c.hwid.is_(None))
+        .values(
+            hwid={
+                "enabled": True,
+                "forced": False,
+                "fallback_limit": None,
+                "min_limit": None,
+                "max_limit": None,
+            }
+        )
+    )
+
+    with op.batch_alter_table("admin_roles", schema=None) as batch_op:
+        batch_op.alter_column("hwid", existing_type=sa.JSON(), type_=sa.JSON(), nullable=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("admin_roles", schema=None) as batch_op:
+        batch_op.drop_column("hwid")

app/db/models.py

@@ -886,6 +886,7 @@ class AdminRole(Base):
     limits: Mapped[Dict] = mapped_column(PostgresJSONB, default_factory=dict)
     features: Mapped[Dict] = mapped_column(PostgresJSONB, default_factory=dict)
     access: Mapped[Dict] = mapped_column(PostgresJSONB, default_factory=dict)
+    hwid: Mapped[Dict] = mapped_column(PostgresJSONB, default_factory=dict)
     disabled_when_limited: Mapped[bool] = mapped_column(default=False, server_default="0")
     disable_users_when_limited: Mapped[bool] = mapped_column(default=True, server_default="1")
     created_at: Mapped[dt] = mapped_column(DateTime(timezone=True), default_factory=lambda: dt.now(tz.utc), init=False)

app/models/admin.py

@@ -10,6 +10,7 @@
 
 from app.db.models import AdminStatus
 from app.models.admin_role import RoleAccess, RoleFeatures, RoleLimits, RolePermissions
+from app.models.settings import HWIDSettings
 from app.models.stats import Period
 from app.utils.helpers import fix_datetime_timezone
 
@@ -58,6 +59,7 @@ class AdminRoleData(BaseModel):
     limits: RoleLimits = Field(default_factory=RoleLimits)
     features: RoleFeatures = Field(default_factory=RoleFeatures)
     access: RoleAccess = Field(default_factory=RoleAccess)
+    hwid: HWIDSettings = Field(default_factory=HWIDSettings)
     disabled_when_limited: bool = False
     disable_users_when_limited: bool = True
 

app/models/admin_role.py

@@ -4,6 +4,7 @@
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
+from app.models.settings import HWIDSettings
 from app.models.validators import ListValidator
 
 
@@ -139,6 +140,7 @@ class AdminRoleBase(BaseModel):
     limits: RoleLimits = Field(default_factory=RoleLimits)
     features: RoleFeatures = Field(default_factory=RoleFeatures)
     access: RoleAccess = Field(default_factory=RoleAccess)
+    hwid: HWIDSettings = Field(default_factory=HWIDSettings)
     disabled_when_limited: bool = False
     disable_users_when_limited: bool = True
 
@@ -155,6 +157,7 @@ class AdminRoleModify(BaseModel):
     limits: RoleLimits | None = None
     features: RoleFeatures | None = None
     access: RoleAccess | None = None
+    hwid: HWIDSettings | None = None
     disabled_when_limited: bool | None = None
     disable_users_when_limited: bool | None = None
 

app/models/settings.py

@@ -288,11 +288,11 @@ def validate_recommended_apps(cls, v: list[Application]) -> list[Application]:
 
 
 class HWIDSettings(BaseModel):
-    enabled: bool = Field(default=False)
+    enabled: bool = Field(default=True)
     forced: bool = Field(default=False)
-    fallback_limit: int = Field(default=0, ge=0)
-    min_limit: int = Field(default=0, ge=0)
-    max_limit: int = Field(default=0, ge=0)
+    fallback_limit: int | None = Field(default=None, ge=0)
+    min_limit: int | None = Field(default=None, ge=0)
+    max_limit: int | None = Field(default=None, ge=0)
 
 
 class General(BaseModel):

app/operation/subscription.py

@@ -15,6 +15,7 @@
 from app.db.models import User
 from app.models.admin import AdminDetails
 from app.models.settings import Application, ConfigFormat, HWIDSettings, SubRule, Subscription as SubSettings
+from app.utils.hwid import resolve_effective_hwid_settings
 from app.models.stats import UserUsageStatsList
 from app.models.subscription import SubscriptionUsageQuery
 from app.models.user import SubscriptionUserResponse, UsersResponseWithInbounds
@@ -281,17 +282,20 @@ async def validate_and_register_hwid(
         db: AsyncSession,
         user_id: int,
         user_hwid_limit: int | None,
+        role_hwid_settings: HWIDSettings | None,
         x_hwid: str | None,
         x_device_os: str | None,
         x_ver_os: str | None,
         x_device_model: str | None,
     ):
-        hwid_conf: HWIDSettings = await hwid_settings()
-        if not hwid_conf.enabled:
+        global_hwid_conf: HWIDSettings = await hwid_settings()
+        effective_hwid_conf = resolve_effective_hwid_settings(global_hwid_conf, role_hwid_settings)
+
+        if effective_hwid_conf is None or not effective_hwid_conf.enabled:
             return
 
         if not x_hwid:
-            if hwid_conf.forced:
+            if effective_hwid_conf.forced:
                 await self.raise_error(message="HWID header required", code=403)
             return
 
@@ -301,7 +305,7 @@ async def validate_and_register_hwid(
             return
 
         # It's a new HWID, check limit
-        limit = user_hwid_limit if user_hwid_limit is not None else hwid_conf.fallback_limit
+        limit = user_hwid_limit if user_hwid_limit is not None else effective_hwid_conf.fallback_limit
         if limit == 0:
             pass  # unlimited
         else:
@@ -360,7 +364,14 @@ async def user_subscription(
             )
         else:
             await self.validate_and_register_hwid(
-                db, db_user.id, db_user.hwid_limit, x_hwid, x_device_os, x_ver_os, x_device_model
+                db,
+                db_user.id,
+                db_user.hwid_limit,
+                db_user.admin.role.hwid if db_user.admin and db_user.admin.role else None,
+                x_hwid,
+                x_device_os,
+                x_ver_os,
+                x_device_model,
             )
             matched_rule = self.detect_client_rule(user_agent, sub_settings.rules)
             client_type = matched_rule.target if matched_rule else None
@@ -443,7 +454,14 @@ async def user_subscription_with_client_type(
         user = await self.validated_user(db_user)
 
         await self.validate_and_register_hwid(
-            db, db_user.id, db_user.hwid_limit, x_hwid, x_device_os, x_ver_os, x_device_model
+            db,
+            db_user.id,
+            db_user.hwid_limit,
+            db_user.admin.role.hwid if db_user.admin and db_user.admin.role else None,
+            x_hwid,
+            x_device_os,
+            x_ver_os,
+            x_device_model,
         )
 
         response_headers = self.create_response_headers(

app/operation/user.py

@@ -55,6 +55,7 @@
 from app.db.models import User, UserStatus, UserTemplate
 from app.models.admin import AdminDetails
 from app.models.proxy import ProxyTable
+from app.models.settings import HWIDSettings
 from app.models.stats import (
     Period,
     UserCountMetric,
@@ -94,7 +95,7 @@
     UserUsageQuery,
     WireGuardPeerIPsReallocateResponse,
 )
-from app.node.sync import remove_user as sync_remove_user, sync_users, sync_user
+from app.node.sync import remove_user as sync_remove_user, sync_user, sync_users
 from app.operation import BaseOperation, OperatorType
 from app.operation.permissions import (
     PermissionDenied,
@@ -105,9 +106,10 @@
     is_scope_all,
 )
 from app.settings import hwid_settings, subscription_settings
+from app.utils.helpers import fix_datetime_timezone
+from app.utils.hwid import resolve_effective_hwid_settings
 from app.utils.jwt import create_subscription_token
 from app.utils.logger import get_logger
-from app.utils.helpers import fix_datetime_timezone
 from app.utils.system import readable_duration, readable_size
 from app.utils.wireguard import (
     build_wireguard_peer_ip_allocator,
@@ -515,10 +517,14 @@ async def _enforce_user_limits(
     async def create_user(
         self, db: AsyncSession, new_user: UserCreate, admin: AdminDetails, *, skip_role_limits: bool = False
     ) -> UserResponse:
-        hwid_conf = await hwid_settings()
+        global_hwid_conf = await hwid_settings()
+        effective_hwid_conf = resolve_effective_hwid_settings(
+            global_hwid_conf,
+            admin.role.hwid if admin.role is not None else None,
+        )
 
         if new_user.hwid_limit is None:
-            new_user.hwid_limit = hwid_conf.fallback_limit
+            new_user.hwid_limit = 0 if effective_hwid_conf is None else (effective_hwid_conf.fallback_limit or 0)
 
         if not skip_role_limits:
             await self._enforce_user_limits(
@@ -534,11 +540,17 @@ async def create_user(
                 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.hwid_limit is not None and not admin.is_owner and effective_hwid_conf is not None:
+            if effective_hwid_conf.min_limit is not None and new_user.hwid_limit < effective_hwid_conf.min_limit:
+                await self.raise_error(
+                    message=f"HWID limit cannot be less than {effective_hwid_conf.min_limit}", code=400, db=db
+                )
+            if effective_hwid_conf.max_limit is not None and effective_hwid_conf.max_limit > 0 and (
+                new_user.hwid_limit > effective_hwid_conf.max_limit or new_user.hwid_limit == 0
+            ):
+                await self.raise_error(
+                    message=f"HWID limit cannot exceed {effective_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)
@@ -625,13 +637,22 @@ async def _prepare_modified_user(
             )
 
         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)
+            global_hwid_conf = await hwid_settings()
+            effective_hwid_conf = resolve_effective_hwid_settings(
+                global_hwid_conf,
+                admin.role.hwid if admin.role is not None else None,
+            )
+            if effective_hwid_conf is not None:
+                if effective_hwid_conf.min_limit is not None and modified_user.hwid_limit < effective_hwid_conf.min_limit:
+                    await self.raise_error(
+                        message=f"HWID limit cannot be less than {effective_hwid_conf.min_limit}", code=400, db=db
+                    )
+                if effective_hwid_conf.max_limit is not None and effective_hwid_conf.max_limit > 0 and (
+                    modified_user.hwid_limit > effective_hwid_conf.max_limit or modified_user.hwid_limit == 0
+                ):
+                    await self.raise_error(
+                        message=f"HWID limit cannot exceed {effective_hwid_conf.max_limit}", code=400, db=db
+                    )
 
         validated_groups = None
         if modified_user.group_ids:

app/utils/hwid.py

@@ -0,0 +1,29 @@
+from app.models.settings import HWIDSettings
+
+
+def resolve_effective_hwid_settings(
+    global_hwid: HWIDSettings, role_hwid: HWIDSettings | dict | None
+) -> HWIDSettings | None:
+    if isinstance(role_hwid, dict):
+        if not role_hwid:
+            role_hwid = None
+        else:
+            try:
+                role_hwid = HWIDSettings.model_validate(role_hwid)
+            except Exception:
+                role_hwid = None
+
+    if role_hwid is None:
+        return global_hwid
+
+    if not role_hwid.enabled:
+        return None
+
+    # enabled=True: None fields inherit from global, explicit values (including 0) override
+    return HWIDSettings(
+        enabled=True,
+        forced=role_hwid.forced,
+        fallback_limit=role_hwid.fallback_limit if role_hwid.fallback_limit is not None else global_hwid.fallback_limit,
+        min_limit=role_hwid.min_limit if role_hwid.min_limit is not None else global_hwid.min_limit,
+        max_limit=role_hwid.max_limit if role_hwid.max_limit is not None else global_hwid.max_limit,
+    )