127 lines
5.2 KiB
Python
127 lines
5.2 KiB
Python
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import bcrypt
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import config, mode
|
|
from .db import get_db
|
|
from .models import User
|
|
|
|
|
|
def hash_password(pw: str) -> str:
|
|
return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode()
|
|
|
|
|
|
def verify_password(pw: str, hashed: str) -> bool:
|
|
try:
|
|
return bcrypt.checkpw(pw.encode(), hashed.encode())
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def make_token(user: User | int, demo: bool = False, rid: int | None = None) -> str:
|
|
"""Access-токен. demo=True — сессия в демо-БД (sub — id в демо-БД, rid — id того же человека в боевой БД,
|
|
None — чисто демонстрационная учётная запись)."""
|
|
exp = datetime.now(timezone.utc) + timedelta(minutes=config.ACCESS_TTL_MIN)
|
|
uid = user if isinstance(user, int) else user.id
|
|
claims = {"sub": str(uid), "exp": exp, "typ": "access", "mode": "demo" if demo else "real"}
|
|
if demo:
|
|
claims.update(rid=rid, ep=mode.State.epoch)
|
|
return jwt.encode(claims, config.SECRET_KEY, algorithm="HS256")
|
|
|
|
|
|
def new_refresh_token() -> tuple[str, str]:
|
|
"""(токен для клиента, sha256-хэш для БД)."""
|
|
raw = secrets.token_urlsafe(48)
|
|
return raw, hash_refresh(raw)
|
|
|
|
|
|
def hash_refresh(raw: str) -> str:
|
|
return hashlib.sha256(raw.encode()).hexdigest()
|
|
|
|
|
|
def current_user(request: Request, db: Session = Depends(get_db)) -> User:
|
|
auth = request.headers.get("authorization", "")
|
|
token = auth[7:] if auth.lower().startswith("bearer ") else None
|
|
if not token:
|
|
raise HTTPException(401, "Требуется авторизация")
|
|
try:
|
|
data = jwt.decode(token, config.SECRET_KEY, algorithms=["HS256"])
|
|
demo = data.get("mode", "real") == "demo"
|
|
rid = data.get("rid")
|
|
if demo:
|
|
# демо-сессия: демо-БД должна существовать, не пересоздаваться с выдачи токена,
|
|
# а человек — всё ещё числиться в демо (администратор мог его вывести)
|
|
if not db.info.get("demo") or data.get("ep") != mode.State.epoch or (rid is not None and not mode.user_in_demo(rid)):
|
|
raise ValueError("mode")
|
|
elif db.info.get("demo") or mode.user_in_demo(int(data["sub"])):
|
|
raise ValueError("mode") # человека перевели в демо — клиент обновит токен и получит демо-сессию
|
|
user = db.get(User, int(data["sub"]))
|
|
except Exception:
|
|
raise HTTPException(401, "Недействительный токен")
|
|
if not user or not user.is_active:
|
|
raise HTTPException(401, "Пользователь не найден или отключён")
|
|
request.state.real_user_id = rid if demo else user.id
|
|
request.state.demo = demo
|
|
return user
|
|
|
|
|
|
def is_engineer(u: User) -> bool:
|
|
return bool(u.position and u.position.kind == "engineer")
|
|
|
|
|
|
def is_manager(u: User) -> bool:
|
|
return bool(u.is_admin or u.is_moderator)
|
|
|
|
|
|
def is_root(u: User) -> bool:
|
|
"""Администратор по умолчанию (логин ADMIN_USERNAME, обычно admin): права администратора портала есть всегда."""
|
|
return bool(u.is_admin and u.username == config.ADMIN_USERNAME)
|
|
|
|
|
|
def is_portal_admin(u: User) -> bool:
|
|
"""Администратор портала: настройки сервера и ответы обратной связи."""
|
|
return bool(u.is_admin and (u.portal_admin or is_root(u)))
|
|
|
|
|
|
def require_portal_admin(u: User = Depends(current_user)) -> User:
|
|
if not is_portal_admin(u):
|
|
raise HTTPException(403, "Доступно только администраторам портала")
|
|
return u
|
|
|
|
|
|
def require_admin(u: User = Depends(current_user)) -> User:
|
|
"""Параметры сервера — только администратор."""
|
|
if not u.is_admin:
|
|
raise HTTPException(403, "Только для администратора")
|
|
return u
|
|
|
|
|
|
def require_manager(u: User = Depends(current_user)) -> User:
|
|
"""Администратор или модератор: пользователи, должности, графики."""
|
|
if not is_manager(u):
|
|
raise HTTPException(403, "Только для администратора или модератора")
|
|
return u
|
|
|
|
|
|
def require_engineer(u: User = Depends(current_user)) -> User:
|
|
"""Инженер, модератор или администратор."""
|
|
if not (is_manager(u) or is_engineer(u)):
|
|
raise HTTPException(403, "Недостаточно прав")
|
|
return u
|
|
|
|
|
|
def is_storekeeper(u: User) -> bool:
|
|
return bool(u.is_admin or u.is_storekeeper)
|
|
|
|
|
|
def require_storekeeper(u: User = Depends(current_user)) -> User:
|
|
"""Администратор склада инструмента или администратор: каталог, склады, выдача, заявки."""
|
|
if not is_storekeeper(u):
|
|
raise HTTPException(403, "Только для администратора склада")
|
|
return u
|