前后端重构权限认证机制，简单化处理权限功能：后端只进行增删改的权限验证

server/common/auth_casbin.py

@@ -0,0 +1,16 @@
+from fastapi import Request, HTTPException
+from ..settings import casbin_enforcer
+
+
+class Authority:
+    def __init__(self, policy: str):
+        self.policy = policy
+
+    def __call__(self, request: Request):
+        if request.state.uid == 1:
+            # uid为1的都跳过，为admin用户
+            return True
+        model, act = self.policy.split(':')
+        if not casbin_enforcer.enforce(f'uid_{request.state.uid}', model, act):
+            print('没有权限')
+            raise HTTPException(status_code=403, detail="没有权限")

server/db.py → server/common/database.py

@@ -1,7 +1,5 @@
 from sqlmodel import create_engine, SQLModel, Session, select
-from .settings import settings
-
-engine = create_engine(settings.DATABASE_URI, future=False)
+from ..settings import engine
 
 
 def init_db():

server/common/security.py

@@ -1,5 +1,7 @@
 from datetime import datetime, timedelta
+from fastapi import Request, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
+from fastapi.security.utils import get_authorization_scheme_param
 from jose import JWTError, jwt
 from passlib.context import CryptContext
 from ..settings import settings
@@ -13,8 +15,31 @@
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
 
-def hash_password(password):
-    return pwd_context.hash(password)
+def auth_check(request: Request):
+    """
+    检查是否有token信息，并在request.state中添加uid值
+    :param request:
+    :return:
+    """
+    for url in settings.NO_VERIFY_URL:
+        if url == request.url.path.lower():
+            print(f"{request.url.path} 在白名单中，不需要权限验证")
+            return True
+    authorization: str = request.headers.get("Authorization")
+    schema, param = get_authorization_scheme_param(authorization)
+    if not authorization or schema.lower() != "bearer":
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authenticated")
+
+    try:
+        playload = jwt.decode(param, settings.SECRET_KEY, settings.ALGORITHM)
+    except jwt.ExpiredSignatureError as e:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
+    except JWTError as e:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
+
+    uid = playload.get('uid')
+    # 在Request对象中设置用户对象，这样在其他地方就能通过request.state.uid获取当前用户id了
+    request.state.uid = uid
 
 
 def create_access_token(data):
@@ -30,27 +55,5 @@ def create_access_token(data):
     else:
         expire = datetime.utcnow() + timedelta(minutes=15)
     to_encode.update({"exp": expire})
-    encoded_jwt = token_encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
     return encoded_jwt
-
-
-def token_encode(to_encode, secret_key, algorithm):
-    """
-    token加密
-    :param to_encode:
-    :param secret_key:
-    :param algorithm:
-    :return:
-    """
-    return jwt.encode(to_encode, secret_key, algorithm)
-
-
-def token_decode(token, secret_key=settings.SECRET_KEY, algorithm=settings.ALGORITHM):
-    """
-    token解密
-    :param token:
-    :param secret_key:
-    :param algorithm:
-    :return:
-    """
-    return jwt.decode(token, secret_key, algorithm)

server/crud/internal/__init__.py

@@ -1,4 +1,3 @@
 from .user import user
 from .roles import role
 from .menu import menu
-from .sysapi import api

server/crud/internal/roles.py

@@ -2,7 +2,7 @@
 from sqlmodel import select, Session
 from ...models.internal import Role, Menu, RoleMenu
 from ..base import CRUDBase
-from ...dependencies import casbin_enforcer
+from ...settings import casbin_enforcer
 
 
 class CRUDRole(CRUDBase[Role]):
@@ -49,11 +49,10 @@ def update_menus(self, session: Session, db_obj: Role, menus: List[int]):
         casbin_enforcer.delete_permissions_for_user(f'role_{db_obj.id}')
         print(db_menus)
         for menu in db_menus:
-            if len(menu.apis) == 0:
-                continue
-            for api in menu.apis:
-                print(f'增加权限:role_{db_obj.id},{api.path},{api.method}')
-                casbin_enforcer.add_permission_for_user(f'role_{db_obj.id}', api.path, api.method)
+            if menu.auth is not None:
+                model, act = menu.auth.split(':')
+                print(f'增加权限:role_{db_obj.id},{model},{act}')
+                casbin_enforcer.add_permission_for_user(f'role_{db_obj.id}', model, act)
         session.add(db_obj)
         session.commit()
         session.refresh(db_obj)

server/db_init.py

@@ -1,2 +1,18 @@
-from sqlmodel import create_engine
-from .sql.models import *
+from sqlmodel import Session
+from .db import init_db, engine
+from .models.role import Role
+from .models.user import User
+
+
+def main():
+    init_db()
+    with Session(engine) as session:
+        admin_role = Role(name='admin', description='Admin管理员组', enable=1)
+        admin = User(name='admin', enable=1, password='1111111')
+        admin.roles.append(admin_role)
+        session.add(admin)
+        session.commit()
+
+
+if __name__ == '__main__':
+    main()

server/main.py

@@ -1,18 +1,16 @@
 from sqlmodel import Session
 from fastapi import FastAPI, Depends
-from .routers.internal import api, login, user, menu, roles
-from .dependencies import check_permission
-from .models.internal.api import Api
-from .db import engine, get_or_create
+from .routers.internal import login, user, menu, roles
+from .common.security import auth_check
+from .settings import engine
 
-app = FastAPI()
+app = FastAPI(dependencies=[Depends(auth_check)])
 
 # 不执行check_permission的，表示不需要权限验证
 app.include_router(login.router, tags=['用户登录'])
-app.include_router(user.router, tags=['用户管理'], dependencies=[Depends(check_permission)])
-app.include_router(menu.router, tags=['菜单管理'], dependencies=[Depends(check_permission)])
-app.include_router(roles.router, tags=['角色管理'], dependencies=[Depends(check_permission)])
-app.include_router(api.router, tags=['接口管理'], dependencies=[Depends(check_permission)])
+app.include_router(user.router, tags=['用户管理'])
+app.include_router(menu.router, tags=['菜单管理'])
+app.include_router(roles.router, tags=['角色管理'])
 
 
 @app.on_event("startup")
@@ -23,24 +21,6 @@ def startup():
     """
     print('服务启动后执行服务')
     print('API接口更新')
-    with Session(engine) as session:
-        for url, value in app.openapi()['paths'].items():
-            if 'get' in value:
-                method = 'get'
-                tags = value[method]['tags'][0]
-                get_or_create(session, Api, tags=tags, path=url, method='GET', summary=value[method]['summary'])
-            if 'post' in value:
-                method = 'post'
-                tags = value[method]['tags'][0]
-                get_or_create(session, Api, tags=tags, path=url, method='POST', summary=value[method]['summary'])
-            if 'put' in value:
-                method = 'put'
-                tags = value[method]['tags'][0]
-                get_or_create(session, Api, tags=tags, path=url, method='PUT', summary=value[method]['summary'])
-            if 'delete' in value:
-                method = 'delete'
-                tags = value[method]['tags'][0]
-                get_or_create(session, Api, tags=tags, path=url, method='DELETE', summary=value[method]['summary'])
 
 
 @app.on_event("shutdown")

server/models/internal/__init__.py

@@ -1,4 +1,3 @@
 from .user import User
 from .menu import Menu
 from .role import Role, RoleMenu
-from .api import Api

server/models/internal/menu.py

@@ -1,42 +1,29 @@
-from typing import Optional, List, Any, TYPE_CHECKING
+from typing import Optional, List
 from sqlmodel import SQLModel, Field, Relationship, Column, Boolean, Integer
-from .relationships import RoleMenu, MenuApis
+from .relationships import RoleMenu
 
 
 class MenuBase(SQLModel):
     id: Optional[int] = Field(sa_column=Column('id', Integer, primary_key=True, autoincrement=True))
     name: str = Field(max_length=20, sa_column_kwargs={'comment': '菜单名'})
     path: Optional[str] = Field(max_length=100, sa_column_kwargs={'comment': '路径'})
     component: Optional[str] = Field(max_length=50, sa_column_kwargs={'comment': '组件'})
+    auth: Optional[str] = Field(max_length=50, sa_column_kwargs={'comment': '授权标识'})
     type: str = Field(max_length=10, sa_column_kwargs={'comment': '类型'})
     parent_id: Optional[int] = Field(sa_column_kwargs={'comment': '父级ID'})
     enable: bool = Field(default=True, sa_column=Column(Boolean, comment='启用'))
 
 
 class Menu(MenuBase, table=True):
     roles: List["Role"] = Relationship(back_populates="menus", link_model=RoleMenu)
-    apis: List['Api'] = Relationship(back_populates='menus', link_model=MenuApis)
     # apis: List['Api'] = Relationship(back_populates="menus", link_model=MenuApi)
 
 
-class MenuRead(MenuBase):
-    apis: List['Api'] = []
-
-
 class MenusWithChild(MenuBase):
-    apis: List['Api'] = []
     children: List['MenusWithChild'] = []
 
 
-class MenuWithUpdate(MenuBase):
-    # 更新菜单信息
-    # id: Optional[int]
-    apis: List[int] = []
-
-
 # 底部导入，且延迟注释
 from .role import Role
-from .api import Api
 
 MenusWithChild.update_forward_refs()
-MenuRead.update_forward_refs()

server/models/internal/relationships.py

@@ -14,8 +14,3 @@ class UserRole(SQLModel, table=True):
     user_id: int = Field(foreign_key="user.id", primary_key=True)
     role_id: int = Field(foreign_key="roles.id", primary_key=True)
 
-
-class MenuApis(SQLModel, table=True):
-    __tablename__ = 'menu_apis'
-    menu_id: int = Field(foreign_key='menu.id', primary_key=True)
-    api_id: int = Field(foreign_key='sys_api.id', primary_key=True)