diff --git a/backend/app/api/v1/endpoints/instance_group.py b/backend/app/api/v1/endpoints/instance_group.py
index 794ad42..7e5d5e5 100644
--- a/backend/app/api/v1/endpoints/instance_group.py
+++ b/backend/app/api/v1/endpoints/instance_group.py
@@ -25,7 +25,11 @@ def _serialize_group(group: InstanceGroup, instance_count: int = 0, user_count:
         "name": group.name,
         "description": group.description,
         "db_type": group.db_type,
+        # Legacy field
         "permissions": group.get_permissions(),
+        # New fields
+        "access_type": group.access_type,
+        "allowed_databases": group.get_allowed_databases(),
         "is_active": group.is_active,
         "created_at": group.created_at,
         "updated_at": group.updated_at,
@@ -95,6 +99,14 @@ async def create_instance_group(
         name=data.name,
         description=data.description,
         db_type=data.db_type,
+        # New fields
+        access_type=data.access_type,
+        allowed_databases=(
+            json.dumps(data.allowed_databases, ensure_ascii=False)
+            if data.allowed_databases
+            else None
+        ),
+        # Legacy field (kept for compat)
         permissions=json.dumps(data.permissions, ensure_ascii=False) if data.permissions else None,
         is_active=data.is_active,
     )
@@ -145,6 +157,7 @@ async def update_instance_group(
         if existing.scalar_one_or_none():
             raise HTTPException(status_code=400, detail="权限组名称已存在")
 
+    # Handle legacy permissions field
     if "permissions" in update_data:
         update_data["permissions"] = (
             json.dumps(update_data["permissions"], ensure_ascii=False)
@@ -152,6 +165,14 @@ async def update_instance_group(
             else None
         )
 
+    # Handle new allowed_databases field
+    if "allowed_databases" in update_data:
+        update_data["allowed_databases"] = (
+            json.dumps(update_data["allowed_databases"], ensure_ascii=False)
+            if update_data["allowed_databases"]
+            else None
+        )
+
     for key, value in update_data.items():
         setattr(group, key, value)
     await db.commit()
diff --git a/backend/app/models/instance_group.py b/backend/app/models/instance_group.py
index 7f67131..61da159 100644
--- a/backend/app/models/instance_group.py
+++ b/backend/app/models/instance_group.py
@@ -1,4 +1,8 @@
+from __future__ import annotations
+
 import json
+from typing import List, Optional
+
 from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime, ForeignKey, Table, func
 from sqlalchemy.orm import relationship
 
@@ -10,7 +14,7 @@ user_groups = Table(
     Base.metadata,
     Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
     Column("group_id", Integer, ForeignKey("instance_groups.id", ondelete="CASCADE"), primary_key=True),
-    Column("permissions", Text, nullable=True),  # JSON: 覆盖组默认权限
+    Column("permissions", Text, nullable=True),  # JSON: 覆盖组默认权限 (legacy, kept for compat)
 )
 
 
@@ -21,7 +25,9 @@ class InstanceGroup(Base):
     name = Column(String(100), nullable=False, unique=True)
     description = Column(Text, nullable=True)
     db_type = Column(String(20), nullable=False, default="all")  # mysql/redis/pg/mongo/all
-    permissions = Column(Text, nullable=True)  # JSON: ["query:select", ...]
+    permissions = Column(Text, nullable=True)  # JSON: ["query:select", ...] (legacy, kept for compat)
+    access_type = Column(String(20), nullable=True, default="readonly")  # readonly / datachange / ddl
+    allowed_databases = Column(Text, nullable=True)  # JSON array of database names; empty = all databases
     is_active = Column(Boolean, default=True)
     created_at = Column(DateTime, server_default=func.now())
     updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
@@ -30,6 +36,8 @@ class InstanceGroup(Base):
     instances = relationship("Instance", back_populates="group", lazy="selectin")
     users = relationship("User", secondary=user_groups, back_populates="groups", lazy="selectin")
 
+    # ---- Legacy helpers (kept for backward compatibility) ----
+
     def get_permissions(self) -> list:
         if self.permissions:
             try:
@@ -40,3 +48,21 @@ class InstanceGroup(Base):
 
     def set_permissions(self, perms: list):
         self.permissions = json.dumps(perms, ensure_ascii=False) if perms else None
+
+    # ---- New access_type / allowed_databases helpers ----
+
+    def get_allowed_databases(self) -> list[str]:
+        """返回允许的数据库名列表，空列表表示关联实例的所有库。"""
+        if self.allowed_databases:
+            try:
+                return json.loads(self.allowed_databases)
+            except (json.JSONDecodeError, TypeError):
+                return []
+        return []
+
+    def set_allowed_databases(self, databases: list[str] | None):
+        self.allowed_databases = (
+            json.dumps(databases, ensure_ascii=False) if databases else None
+        )
+
+    VALID_ACCESS_TYPES = {"readonly", "datachange", "ddl"}
diff --git a/backend/app/schemas/instance_group.py b/backend/app/schemas/instance_group.py
index 80c159a..f763654 100644
--- a/backend/app/schemas/instance_group.py
+++ b/backend/app/schemas/instance_group.py
@@ -1,12 +1,22 @@
 from datetime import datetime
 from typing import Optional, List
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 class InstanceGroupCreate(BaseModel):
     name: str = Field(..., min_length=1, max_length=100)
     description: Optional[str] = None
     db_type: str = Field(default="all", pattern=r"^(mysql|redis|pg|mongo|all)$")
+    access_type: str = Field(
+        default="readonly",
+        pattern=r"^(readonly|datachange|ddl)$",
+        description="权限组类型: readonly(只读查询) / datachange(数据变更) / ddl(DDL变更)",
+    )
+    allowed_databases: Optional[List[str]] = Field(
+        default=None,
+        description="允许访问的数据库名列表，为空表示关联实例的所有库",
+    )
+    # Legacy field kept for backward compatibility
     permissions: Optional[List[str]] = None
     is_active: bool = True
 
@@ -15,6 +25,16 @@ class InstanceGroupUpdate(BaseModel):
     name: Optional[str] = Field(default=None, min_length=1, max_length=100)
     description: Optional[str] = None
     db_type: Optional[str] = Field(default=None, pattern=r"^(mysql|redis|pg|mongo|all)$")
+    access_type: Optional[str] = Field(
+        default=None,
+        pattern=r"^(readonly|datachange|ddl)$",
+        description="权限组类型: readonly(只读查询) / datachange(数据变更) / ddl(DDL变更)",
+    )
+    allowed_databases: Optional[List[str]] = Field(
+        default=None,
+        description="允许访问的数据库名列表，为空表示关联实例的所有库",
+    )
+    # Legacy field kept for backward compatibility
     permissions: Optional[List[str]] = None
     is_active: Optional[bool] = None
 
@@ -24,7 +44,11 @@ class InstanceGroupResponse(BaseModel):
     name: str
     description: Optional[str] = None
     db_type: str
+    # Legacy field
     permissions: Optional[List[str]] = None
+    # New fields
+    access_type: Optional[str] = None
+    allowed_databases: Optional[List[str]] = None
     is_active: bool
     created_at: Optional[datetime] = None
     updated_at: Optional[datetime] = None
diff --git a/backend/app/services/role_permissions.py b/backend/app/services/role_permissions.py
new file mode 100644
index 0000000..c203397
--- /dev/null
+++ b/backend/app/services/role_permissions.py
@@ -0,0 +1,288 @@
+"""角色权限服务 — 基于角色的权限查询。
+
+权限逻辑:
+- admin/dba/auditor 权限由角色硬编码决定，不需要查权限组
+- developer 权限由关联的权限组决定
+- 工单操作权限完全由角色决定
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Dict, List, Optional, Set
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.user import User
+from app.models.instance import Instance
+from app.models.instance_group import InstanceGroup, user_groups
+
+logger = logging.getLogger(__name__)
+
+# ============================================================
+# 角色权限映射（硬编码）
+# ============================================================
+# access_type 值:
+#   "all"    — 全部实例全部库
+#   "groups" — 由权限组决定
+#   None     — 无权限
+# ticket 值: list of allowed ticket actions
+ROLE_PERMISSIONS: Dict[str, dict] = {
+    "admin": {
+        "readonly": "all",
+        "datachange": "all",
+        "ddl": "all",
+        "ticket": ["create", "approve", "reject", "execute", "cancel"],
+    },
+    "dba": {
+        "readonly": "all",
+        "datachange": "all",
+        "ddl": "all",
+        "ticket": ["approve", "reject", "execute"],
+    },
+    "auditor": {
+        "readonly": "all",
+        "datachange": None,
+        "ddl": None,
+        "ticket": ["approve", "reject"],
+    },
+    "developer": {
+        "readonly": "groups",
+        "datachange": "groups",
+        "ddl": "groups",
+        "ticket": ["create", "cancel_own"],
+    },
+    "guest": {
+        "readonly": None,
+        "datachange": None,
+        "ddl": None,
+        "ticket": [],
+    },
+}
+
+
+def _get_role_config(role: str) -> dict:
+    """获取角色配置，未知角色按 guest 处理。"""
+    return ROLE_PERMISSIONS.get(role, ROLE_PERMISSIONS["guest"])
+
+
+# ============================================================
+# Public API
+# ============================================================
+
+
+async def get_user_permissions(db: AsyncSession, user: User) -> dict:
+    """返回用户的实际权限概览。
+
+    Returns:
+        {
+            "role": "developer",
+            "access": {
+                "readonly": "all" | "groups" | None,
+                "datachange": "all" | "groups" | None,
+                "ddl": "all" | "groups" | None,
+            },
+            "ticket": ["create", ...],
+            "groups": [  # 仅 developer 类型有详情
+                {
+                    "group_id": 1,
+                    "group_name": "xxx",
+                    "access_type": "readonly",
+                    "instance_ids": [1, 2],
+                    "allowed_databases": ["db1", "db2"],
+                },
+                ...
+            ],
+        }
+    """
+    role_config = _get_role_config(user.role)
+    result = {
+        "role": user.role,
+        "access": {
+            "readonly": role_config["readonly"],
+            "datachange": role_config["datachange"],
+            "ddl": role_config["ddl"],
+        },
+        "ticket": role_config["ticket"],
+        "groups": [],
+    }
+
+    # 如果角色权限由权限组决定，查询用户关联的权限组
+    has_group_access = any(
+        v == "groups" for v in role_config.values() if isinstance(v, str)
+    )
+    if has_group_access:
+        groups_detail = await _get_user_groups_detail(db, user)
+        result["groups"] = groups_detail
+
+    return result
+
+
+async def can_access_database(
+    db: AsyncSession,
+    user: User,
+    instance_id: int,
+    database: str,
+    access_type: str,
+) -> bool:
+    """判断用户是否可以对指定实例的指定库执行指定类型的操作。
+
+    Args:
+        db: 数据库会话
+        user: 用户对象
+        instance_id: 实例 ID
+        database: 数据库名
+        access_type: "readonly" / "datachange" / "ddl"
+
+    Returns:
+        True 如果允许
+    """
+    if access_type not in ("readonly", "datachange", "ddl"):
+        return False
+
+    role_config = _get_role_config(user.role)
+    access_value = role_config.get(access_type)
+
+    # "all" — 全部实例全部库
+    if access_value == "all":
+        return True
+
+    # None — 无权限
+    if access_value is None:
+        return False
+
+    # "groups" — 由权限组决定
+    if access_value == "groups":
+        return await _check_group_access(db, user, instance_id, database, access_type)
+
+    return False
+
+
+def can_ticket_action(user: User, action: str) -> bool:
+    """判断用户是否可以执行指定工单操作（纯角色判断，不需要查 DB）。
+
+    Args:
+        user: 用户对象
+        action: "create" / "approve" / "reject" / "execute" / "cancel" / "cancel_own"
+
+    Returns:
+        True 如果允许
+    """
+    role_config = _get_role_config(user.role)
+    ticket_perms = role_config.get("ticket", [])
+
+    # cancel_own 是一个特殊权限，caller 需要额外验证工单申请人
+    return action in ticket_perms
+
+
+async def get_accessible_instances(
+    db: AsyncSession,
+    user: User,
+) -> List[Instance]:
+    """获取用户可访问的实例列表。
+
+    - admin/dba/auditor（readonly=all）: 返回所有实例
+    - developer: 返回其权限组关联的实例
+    """
+    role_config = _get_role_config(user.role)
+
+    # 判断是否有 "all" 权限（至少有一个 access_type 为 all）
+    has_all = any(v == "all" for v in role_config.values() if isinstance(v, str))
+    if has_all:
+        result = await db.execute(select(Instance))
+        return list(result.scalars().all())
+
+    # developer: 查询权限组关联的实例
+    group_ids = await _get_user_group_ids(db, user)
+    if not group_ids:
+        return []
+
+    result = await db.execute(
+        select(Instance).where(Instance.group_id.in_(group_ids))
+    )
+    return list(result.scalars().all())
+
+
+# ============================================================
+# Internal helpers
+# ============================================================
+
+
+async def _get_user_group_ids(db: AsyncSession, user: User) -> List[int]:
+    """获取用户关联的所有权限组 ID。"""
+    result = await db.execute(
+        select(user_groups.c.group_id).where(user_groups.c.user_id == user.id)
+    )
+    return [row[0] for row in result.all()]
+
+
+async def _get_user_groups_detail(db: AsyncSession, user: User) -> List[dict]:
+    """获取用户关联权限组的详细信息。"""
+    group_ids = await _get_user_group_ids(db, user)
+    if not group_ids:
+        return []
+
+    result = await db.execute(
+        select(InstanceGroup).where(InstanceGroup.id.in_(group_ids))
+    )
+    groups = result.scalars().all()
+
+    details = []
+    for g in groups:
+        instance_ids = [inst.id for inst in (g.instances or [])]
+        details.append({
+            "group_id": g.id,
+            "group_name": g.name,
+            "access_type": g.access_type,
+            "instance_ids": instance_ids,
+            "allowed_databases": g.get_allowed_databases(),
+        })
+    return details
+
+
+async def _check_group_access(
+    db: AsyncSession,
+    user: User,
+    instance_id: int,
+    database: str,
+    access_type: str,
+) -> bool:
+    """检查用户通过权限组是否能访问指定实例+库。"""
+    # 获取实例信息
+    inst_result = await db.execute(
+        select(Instance).where(Instance.id == instance_id)
+    )
+    instance = inst_result.scalar_one_or_none()
+    if not instance:
+        return False
+
+    if instance.group_id is None:
+        # 公开实例，任何已登录用户都可以 readonly
+        return access_type == "readonly"
+
+    # 检查用户是否属于该实例所在的权限组
+    group_ids = await _get_user_group_ids(db, user)
+    if instance.group_id not in group_ids:
+        return False
+
+    # 获取权限组
+    group_result = await db.execute(
+        select(InstanceGroup).where(InstanceGroup.id == instance.group_id)
+    )
+    group = group_result.scalar_one_or_none()
+    if not group:
+        return False
+
+    # 检查 access_type 是否匹配
+    if group.access_type != access_type:
+        return False
+
+    # 检查 allowed_databases
+    allowed_dbs = group.get_allowed_databases()
+    if not allowed_dbs:
+        # 空列表 = 该实例所有库
+        return True
+
+    return database in allowed_dbs