Spaces:
Sleeping
Sleeping
File size: 4,965 Bytes
2f4298a 2f0d9d4 2f4298a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 |
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from database import get_db
from models import User, UserStats
from schemas import TokenData
from config import get_settings
settings = get_settings()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Проверка пароля"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""Хеширование пароля"""
# Обрезаем пароль до 72 байт для bcrypt
if len(password.encode('utf-8')) > 72:
password = password[:72]
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
"""Создание JWT токена"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def authenticate_user(db: Session, username: str, password: str):
"""Аутентификация пользователя"""
user = db.query(User).filter(User.username == username).first()
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
) -> User:
"""Получение текущего пользователя из токена"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = db.query(User).filter(User.username == token_data.username).first()
if user is None:
raise credentials_exception
return user
async def get_current_active_user(
current_user: User = Depends(get_current_user)
) -> User:
"""Проверка активности пользователя"""
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
async def get_current_admin_user(
current_user: User = Depends(get_current_active_user)
) -> User:
"""Проверка прав администратора"""
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
def create_user(db: Session, username: str, email: str, password: str, is_admin: bool = False) -> User:
"""Создание нового пользователя"""
# Проверка существования пользователя
if db.query(User).filter(User.username == username).first():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered"
)
if db.query(User).filter(User.email == email).first():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered"
)
# Создание пользователя
hashed_password = get_password_hash(password)
user = User(
username=username,
email=email,
hashed_password=hashed_password,
is_admin=is_admin
)
db.add(user)
db.commit()
db.refresh(user)
# Создание статистики пользователя
user_stats = UserStats(user_id=user.id)
db.add(user_stats)
db.commit()
return user
def init_admin_user(db: Session):
"""Инициализация администратора при первом запуске"""
admin = db.query(User).filter(User.username == settings.ADMIN_USERNAME).first()
if not admin:
create_user(
db=db,
username=settings.ADMIN_USERNAME,
email=settings.ADMIN_EMAIL,
password=settings.ADMIN_PASSWORD,
is_admin=True
)
print(f"Admin user created: {settings.ADMIN_USERNAME}")
|