phase0-final-rls-rbac-complete

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)