admin/security.py - master - dmitry.kovalenkov/webcc

admin/security.py
67 строк3 KB
dmkovalenkov
update
27 апр 2026, 21:13
27 апр 2026, 21:13e6bfd88
import httpx
from functools import lru_cache
from jose import jwt, JWTError
from fastapi import HTTPException, Security, Depends, status
from fastapi.security import HTTPBearer
import config
from rbac import resolve_permissions

bearer_scheme = HTTPBearer(auto_error=False)

# ── Публичный ключ Auth-сервиса ─────────────────────────────────────
_public_key: str | None = None


def load_public_key():
    global _public_key
    resp = httpx.get(f"{config.AUTH_SERVICE_URL}/auth/public-key", timeout=5)
    resp.raise_for_status()
    _public_key = resp.json()["public_key"]


# ── Базовая зависимость: распаковать JWT ────────────────────────────
def get_current_user(credentials=Security(bearer_scheme)):
    if credentials is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
    if _public_key is None:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Public key not loaded")
    try:
        payload = jwt.decode(credentials.credentials, _public_key, algorithms=["RS256"])
        return payload
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")


# ── Фабрика зависимостей: проверка конкретного права ────────────────
def require_permission(permission: str):
    """
    Использование:
        @router.get("/groups", dependencies=[Depends(require_permission("groups:read"))])
    или
        payload = Depends(require_permission("groups:read"))
    """
    def _check(payload: dict = Depends(get_current_user)):
        role = payload.get("role", "")
        perms = resolve_permissions(role)
        if permission not in perms:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Permission '{permission}' required",
            )
        return payload
    _check.__name__ = f"require_{permission.replace(':', '_')}"
    return _check


# ── Фабрика: любое из перечисленных прав (OR) ───────────────────────
def require_any_permission(*permissions: str):
    def _check(payload: dict = Depends(get_current_user)):
        role = payload.get("role", "")
        perms = resolve_permissions(role)
        if not perms.intersection(permissions):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"One of permissions required: {permissions}",
            )
        return payload
    return _check