139 lines
4.7 KiB
Python
139 lines
4.7 KiB
Python
|
"""鉴权依赖：从 Authorization header 解析当前用户，提供权限校验。"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from fastapi import Depends, HTTPException, Query, Request, status
|
|||
|
|
|
|||
|
|
from app.db.platform_store import get_platform_store
|
|||
|
|
|
|||
|
|
# 无需鉴权的路径前缀（健康检查、登录等）
|
|||
|
|
PUBLIC_PATHS = ("/health", "/login", "/system-info")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _extract_token(request: Request) -> str | None:
|
|||
|
|
"""从 Authorization header 提取 token（格式: Bearer platform-token-{user_id}）。"""
|
|||
|
|
auth = request.headers.get("Authorization", "")
|
|||
|
|
token = auth.replace("Bearer ", "").strip()
|
|||
|
|
if token.startswith("platform-token-"):
|
|||
|
|
return token[len("platform-token-"):]
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_current_user(request: Request) -> dict[str, Any]:
|
|||
|
|
"""
|
|||
|
|
FastAPI 依赖：解析当前登录用户。
|
|||
|
|
- 公开路径（/health, /login 等）直接放行，返回匿名用户。
|
|||
|
|
- 无 token 或 token 无效时抛 401。
|
|||
|
|
- admin 用户标记为超级管理员，拥有全部权限。
|
|||
|
|
"""
|
|||
|
|
path = request.url.path
|
|||
|
|
# 去掉路由前缀后判断
|
|||
|
|
for prefix in PUBLIC_PATHS:
|
|||
|
|
if path.endswith(prefix):
|
|||
|
|
return {"id": None, "username": "anonymous", "role": "viewer", "permissions": [], "protected": False}
|
|||
|
|
|
|||
|
|
user_id = _extract_token(request)
|
|||
|
|
if not user_id:
|
|||
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing or invalid token")
|
|||
|
|
|
|||
|
|
store = get_platform_store()
|
|||
|
|
for u in store.users():
|
|||
|
|
if u.get("id") == user_id:
|
|||
|
|
return u
|
|||
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def require_admin(current_user: dict[str, Any] = Depends(get_current_user)) -> dict[str, Any]:
|
|||
|
|
"""FastAPI 依赖：要求当前用户是管理员（role=admin 或 protected）。"""
|
|||
|
|
if current_user.get("role") == "admin" or current_user.get("protected"):
|
|||
|
|
return current_user
|
|||
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin permission required")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def is_admin(user: dict[str, Any]) -> bool:
|
|||
|
|
"""判断用户是否为管理员（admin 角色或 protected 标记）。"""
|
|||
|
|
return user.get("role") == "admin" or user.get("protected", False)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def has_resource_access(
|
|||
|
|
resource_type: str,
|
|||
|
|
resource_id: str,
|
|||
|
|
user: dict[str, Any],
|
|||
|
|
permission: str = "read",
|
|||
|
|
) -> bool:
|
|||
|
|
"""
|
|||
|
|
检查用户对某资源是否有指定权限。
|
|||
|
|
- admin/protected 用户直接放行（旁路）。
|
|||
|
|
- 其他用户检查 acls 表中是否有对应授权。
|
|||
|
|
"""
|
|||
|
|
if user.get("role") == "admin" or user.get("protected"):
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
store = get_platform_store()
|
|||
|
|
acls = store.get_acl(resource_type, resource_id)
|
|||
|
|
user_id = user.get("id")
|
|||
|
|
user_role = user.get("role")
|
|||
|
|
|
|||
|
|
for entry in acls:
|
|||
|
|
# 按 user 授权
|
|||
|
|
if entry.get("principal_type") == "user" and entry.get("principal_id") == user_id:
|
|||
|
|
if _permission_covers(entry.get("permission"), permission):
|
|||
|
|
return True
|
|||
|
|
# 按 role 授权
|
|||
|
|
if entry.get("principal_type") == "role" and entry.get("principal_id") == user_role:
|
|||
|
|
if _permission_covers(entry.get("permission"), permission):
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _permission_covers(granted: str | None, required: str) -> bool:
|
|||
|
|
"""权限覆盖判断：write/execute 覆盖 read；admin 覆盖一切。"""
|
|||
|
|
if not granted:
|
|||
|
|
return False
|
|||
|
|
if granted == "admin":
|
|||
|
|
return True
|
|||
|
|
if granted == required:
|
|||
|
|
return True
|
|||
|
|
# write 覆盖 read
|
|||
|
|
if required == "read" and granted in ("write", "execute"):
|
|||
|
|
return True
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def filter_accessible_resource_ids(
|
|||
|
|
resource_type: str,
|
|||
|
|
all_ids: list[str],
|
|||
|
|
user: dict[str, Any],
|
|||
|
|
) -> list[str]:
|
|||
|
|
"""
|
|||
|
|
从全部资源 ID 中过滤出当前用户可访问的 ID 列表。
|
|||
|
|
- admin 直接返回全部。
|
|||
|
|
- 普通用户查 acls 表取交集。
|
|||
|
|
"""
|
|||
|
|
if user.get("role") == "admin" or user.get("protected"):
|
|||
|
|
return all_ids
|
|||
|
|
|
|||
|
|
if not all_ids:
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
store = get_platform_store()
|
|||
|
|
user_id = user.get("id")
|
|||
|
|
user_role = user.get("role")
|
|||
|
|
|
|||
|
|
# 查询该用户在该资源类型下有 read 权限的所有 resource_id
|
|||
|
|
with store.connect() as conn:
|
|||
|
|
rows = conn.execute(
|
|||
|
|
"""
|
|||
|
|
SELECT DISTINCT resource_id FROM acls
|
|||
|
|
WHERE resource_type=? AND (
|
|||
|
|
(principal_type='user' AND principal_id=?)
|
|||
|
|
OR (principal_type='role' AND principal_id=?)
|
|||
|
|
)
|
|||
|
|
""",
|
|||
|
|
(resource_type, user_id, user_role),
|
|||
|
|
).fetchall()
|
|||
|
|
|
|||
|
|
accessible = {r["resource_id"] for r in rows}
|
|||
|
|
return [rid for rid in all_ids if rid in accessible]
|