diff --git a/backend/python/app/auth.py b/backend/python/app/auth.py index c02a28a..c3c759d 100644 --- a/backend/python/app/auth.py +++ b/backend/python/app/auth.py @@ -12,12 +12,35 @@ from fastapi import Depends, HTTPException, status, Request from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from .config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES + +# Phase0: 5-level RBAC +ROLE_HIERARCHY = { + "super_admin": 100, + "admin": 80, + "developer": 50, + "user": 30, + "guest": 10, +} + +def has_role(user_role: str, required: str) -> bool: + return ROLE_HIERARCHY.get(user_role, 0) >= ROLE_HIERARCHY.get(required, 0) + logger = logging.getLogger(__name__) security = HTTPBearer() security_optional = HTTPBearer(auto_error=False) +async def get_current_tenant( + credentials: HTTPAuthorizationCredentials = Depends(security), +): + try: + payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM]) + return payload.get("tenant_id") + except JWTError: + return None + + def hash_password(password: str) -> str: """使用 bcrypt 直接生成密码哈希""" return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") @@ -35,8 +58,10 @@ def verify_password(plain_password: str, hashed_password: str) -> bool: return False -def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: +def create_access_token(data: dict, expires_delta: Optional[timedelta] = None, tenant_id: str = None) -> str: to_encode = data.copy() + if tenant_id: + to_encode["tenant_id"] = tenant_id expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)