110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
import json
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from uuid import uuid4
|
|
|
|
import bcrypt
|
|
|
|
# Настройки JWT (читаются из config/settings)
|
|
from config.settings import settings
|
|
from jose import JWTError, jwt
|
|
from models.user import Token, User, UserCreate
|
|
|
|
SECRET_KEY = settings.JWT_SECRET_KEY
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 дней
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self):
|
|
from config.settings import settings
|
|
|
|
self.data_dir = Path(settings.DATA_DIR)
|
|
self.data_dir.mkdir(exist_ok=True)
|
|
self.users_file = self.data_dir / "users.json"
|
|
self.users = self._load_users()
|
|
|
|
def _load_users(self) -> list[User]:
|
|
if self.users_file.exists():
|
|
with open(self.users_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return [User(**u) for u in data]
|
|
return []
|
|
|
|
def _save_users(self):
|
|
with open(self.users_file, "w", encoding="utf-8") as f:
|
|
json.dump(
|
|
[u.model_dump() for u in self.users], f, ensure_ascii=False, indent=2
|
|
)
|
|
|
|
def verify_password(self, plain_password: str, hashed_password: str) -> bool:
|
|
return bcrypt.checkpw(
|
|
plain_password.encode("utf-8"), hashed_password.encode("utf-8")
|
|
)
|
|
|
|
def get_password_hash(self, password: str) -> str:
|
|
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
|
|
def get_user_by_username(self, username: str) -> Optional[User]:
|
|
for user in self.users:
|
|
if user.username == username:
|
|
return user
|
|
return None
|
|
|
|
def get_user_by_id(self, user_id: str) -> Optional[User]:
|
|
for user in self.users:
|
|
if user.id == user_id:
|
|
return user
|
|
return None
|
|
|
|
def create_user(self, user_create: UserCreate) -> User:
|
|
if self.get_user_by_username(user_create.username):
|
|
raise ValueError("Username already exists")
|
|
|
|
new_user = User(
|
|
id=str(uuid4()),
|
|
username=user_create.username,
|
|
email=user_create.email,
|
|
hashed_password=self.get_password_hash(user_create.password),
|
|
)
|
|
self.users.append(new_user)
|
|
self._save_users()
|
|
|
|
# Создаем папку для проектов пользователя
|
|
user_projects_dir = self.data_dir / "users" / new_user.id / "projects"
|
|
user_projects_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
return new_user
|
|
|
|
def authenticate_user(self, username: str, password: str) -> Optional[User]:
|
|
user = self.get_user_by_username(username)
|
|
if not user or not self.verify_password(password, user.hashed_password):
|
|
return None
|
|
return user
|
|
|
|
def create_access_token(self, data: dict) -> str:
|
|
to_encode = data.copy()
|
|
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
to_encode.update({"exp": expire})
|
|
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
def get_token(self, username: str, password: str) -> Token:
|
|
user = self.authenticate_user(username, password)
|
|
if not user:
|
|
raise ValueError("Invalid username or password")
|
|
|
|
access_token = self.create_access_token(
|
|
data={"sub": user.username, "user_id": user.id}
|
|
)
|
|
return Token(
|
|
access_token=access_token,
|
|
token_type="bearer",
|
|
user_id=user.id,
|
|
username=user.username,
|
|
)
|
|
|
|
|
|
# Singleton
|
|
auth_service = AuthService()
|