91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
import time
|
|
import secrets
|
|
import bcrypt
|
|
from datetime import datetime
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, func
|
|
from app.models.models import User, LoginAttempt, AuditLog
|
|
|
|
BRUTE_FORCE_WINDOW = 180
|
|
BRUTE_FORCE_MAX_ATTEMPTS = 3
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode()[:72], bcrypt.gensalt()).decode()
|
|
|
|
|
|
def verify_password(password: str, hashed: str) -> bool:
|
|
return bcrypt.checkpw(password.encode()[:72], hashed.encode())
|
|
|
|
|
|
def generate_csrf_token() -> str:
|
|
return secrets.token_hex(32)
|
|
|
|
|
|
async def brute_force_check(db: AsyncSession, login: str, ip_address: str) -> bool:
|
|
cutoff = datetime.utcnow().timestamp() - BRUTE_FORCE_WINDOW
|
|
result = await db.execute(
|
|
select(func.count(LoginAttempt.id))
|
|
.where(
|
|
(LoginAttempt.login == login) | (LoginAttempt.ip_address == ip_address),
|
|
LoginAttempt.attempt_time > datetime.fromtimestamp(cutoff),
|
|
)
|
|
)
|
|
count = result.scalar() or 0
|
|
return count >= BRUTE_FORCE_MAX_ATTEMPTS
|
|
|
|
|
|
async def record_attempt(db: AsyncSession, login: str, ip_address: str):
|
|
attempt = LoginAttempt(login=login, ip_address=ip_address)
|
|
db.add(attempt)
|
|
await db.flush()
|
|
|
|
|
|
async def clear_attempts(db: AsyncSession, login: str, ip_address: str):
|
|
from sqlalchemy import delete
|
|
await db.execute(
|
|
delete(LoginAttempt).where(
|
|
(LoginAttempt.login == login) | (LoginAttempt.ip_address == ip_address)
|
|
)
|
|
)
|
|
await db.flush()
|
|
|
|
|
|
async def authenticate(db: AsyncSession, login: str, password: str, ip_address: str) -> dict:
|
|
if await brute_force_check(db, login, ip_address):
|
|
return {"success": False, "error": "Превышен лимит попыток. Подождите 3 минуты."}
|
|
|
|
result = await db.execute(
|
|
select(User).where(User.login == login).limit(1)
|
|
)
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user or not verify_password(password, user.password_hash):
|
|
await record_attempt(db, login, ip_address)
|
|
return {"success": False, "error": "Неверный логин или пароль."}
|
|
|
|
if not user.is_active:
|
|
return {"success": False, "error": "Учётная запись заблокирована."}
|
|
|
|
await clear_attempts(db, login, ip_address)
|
|
await audit_log(db, user.id, "login", "Вход в систему", ip_address)
|
|
|
|
return {
|
|
"success": True,
|
|
"user": {
|
|
"id": user.id,
|
|
"login": user.login,
|
|
"role": user.role,
|
|
"full_name": user.full_name,
|
|
},
|
|
}
|
|
|
|
|
|
async def audit_log(db: AsyncSession, user_id: int | None, action: str, details: str, ip_address: str):
|
|
try:
|
|
log = AuditLog(user_id=user_id, action=action, details=details, ip_address=ip_address)
|
|
db.add(log)
|
|
await db.flush()
|
|
except Exception:
|
|
pass
|