df1856d793
- 产品编辑入口统一走 ProductFormFull(卖点/风格/人群/品牌词全字段); 修复开任务页 <form> 套 <form> 致"编辑产品"报错、改不了、跳回首个产品 - dashboard 入口卡片对齐实际路由: 系统管理(/config) 与 工作配置(/settings) 分开; settings ?tab=products 直达改用挂载后读 URL, 消除 hydration mismatch - 新增用户管理(users API/admin service/改密页) + alembic 022/023/024 - 上线部署: Dockerfile / docker-compose.prod+https / nginx https / .env.example - A8 三套正交叙事(痛点/场景/成分背书) + beige 调色去AI化 + 飞轮 text_import 高权重信号 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""
|
||
app/middleware/workspace_guard.py — 多租户隔离中间件
|
||
所有业务接口强制注入 workspace_id（基石C）。
|
||
读操作信任 JWT；写操作+切换 workspace 查 workspace_members 校验。
|
||
"""
|
||
|
||
import logging
|
||
from typing import Annotated
|
||
|
||
import jwt
|
||
from fastapi import Depends, Header
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.database import get_db
|
||
from app.core.response import raise_forbidden, raise_unauthorized
|
||
from app.core.security import decode_access_token
|
||
from app.models.user import User
|
||
from app.models.workspace import WorkspaceMember
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class CurrentUser:
|
||
"""JWT 解码后的请求上下文，注入到每个路由函数。"""
|
||
|
||
def __init__(
|
||
self,
|
||
user_id: int,
|
||
workspace_id: int,
|
||
role: str,
|
||
):
|
||
self.user_id = user_id
|
||
self.workspace_id = workspace_id
|
||
self.role = role
|
||
|
||
|
||
def _extract_token(authorization: str | None) -> str:
|
||
"""从 Authorization: Bearer <token> 中提取 token。"""
|
||
if not authorization or not authorization.startswith("Bearer "):
|
||
raise_unauthorized("缺少 Authorization header")
|
||
return authorization.split(" ", 1)[1]
|
||
|
||
|
||
async def get_current_user(
|
||
authorization: Annotated[str | None, Header()] = None,
|
||
) -> CurrentUser:
|
||
"""
|
||
FastAPI 依赖：解码 JWT，返回 CurrentUser。
|
||
所有业务路由 Depends(get_current_user)。
|
||
"""
|
||
token = _extract_token(authorization)
|
||
try:
|
||
payload = decode_access_token(token)
|
||
except jwt.ExpiredSignatureError:
|
||
raise_unauthorized("Token 已过期")
|
||
except jwt.PyJWTError:
|
||
raise_unauthorized("Token 无效")
|
||
|
||
return CurrentUser(
|
||
user_id=int(payload["user_id"]),
|
||
workspace_id=int(payload["current_workspace_id"]),
|
||
role=payload["role"],
|
||
)
|
||
|
||
|
||
def require_write_permission(
|
||
current_user: Annotated[CurrentUser, Depends(get_current_user)],
|
||
db: Annotated[Session, Depends(get_db)],
|
||
) -> CurrentUser:
|
||
"""
|
||
写操作依赖：查 workspace_members 校验当前用户确实属于此 workspace。
|
||
读操作用 get_current_user 即可（JWT 加速）。
|
||
"""
|
||
member = (
|
||
db.query(WorkspaceMember)
|
||
.filter(
|
||
WorkspaceMember.workspace_id == current_user.workspace_id,
|
||
WorkspaceMember.user_id == current_user.user_id,
|
||
)
|
||
.first()
|
||
)
|
||
if not member:
|
||
logger.warning(
|
||
"workspace permission denied: user=%s workspace=%s",
|
||
current_user.user_id,
|
||
current_user.workspace_id,
|
||
)
|
||
raise_forbidden("无权限访问此 workspace")
|
||
user = db.query(User).filter(User.id == current_user.user_id).first()
|
||
if user and getattr(user, "must_change_password", False):
|
||
raise_forbidden("首次登录必须先修改密码")
|
||
return current_user
|
||
|
||
|
||
def require_admin(
|
||
current_user: Annotated[CurrentUser, Depends(require_write_permission)],
|
||
) -> CurrentUser:
|
||
"""仅管理员可访问的路由依赖。"""
|
||
if current_user.role != "admin":
|
||
raise_forbidden("需要管理员权限")
|
||
return current_user
|
||
|
||
|
||
def require_supervisor_or_above(
|
||
current_user: Annotated[CurrentUser, Depends(require_write_permission)],
|
||
) -> CurrentUser:
|
||
"""组长及以上（supervisor/admin）。"""
|
||
if current_user.role not in ("supervisor", "admin"):
|
||
raise_forbidden("需要组长或管理员权限")
|
||
return current_user
|