157 lines
4.8 KiB
Python
157 lines
4.8 KiB
Python
|
|
from functions.database import get_db_connection, Usuario
|
||
|
|
from flask_login import login_user, logout_user
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Dict, Optional
|
||
|
|
import pyotp
|
||
|
|
import qrcode
|
||
|
|
import base64
|
||
|
|
from io import BytesIO
|
||
|
|
|
||
|
|
class AuthService:
|
||
|
|
"""Service para operações de autenticação"""
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def autenticar_usuario(email_or_username: str, password: str, otp: str = None) -> Dict:
|
||
|
|
"""Autentica um usuário"""
|
||
|
|
db = get_db_connection()
|
||
|
|
try:
|
||
|
|
# Tenta encontrar o usuário por email ou username
|
||
|
|
user = db.query(Usuario).filter(
|
||
|
|
(Usuario.email == email_or_username) |
|
||
|
|
(Usuario.username == email_or_username)
|
||
|
|
).first()
|
||
|
|
|
||
|
|
if not user or not user.check_password(password):
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': 'Email/usuário ou senha incorretos.'
|
||
|
|
}
|
||
|
|
|
||
|
|
# Verificar OTP se o usuário tiver configurado
|
||
|
|
if user.otp_secret and not otp:
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': 'Código OTP é obrigatório para sua conta.'
|
||
|
|
}
|
||
|
|
|
||
|
|
if user.otp_secret and not user.verify_otp(otp):
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': 'Código OTP inválido.'
|
||
|
|
}
|
||
|
|
|
||
|
|
# Atualizar último login
|
||
|
|
user.ultimo_login = datetime.utcnow()
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
# Fazer login
|
||
|
|
login_user(user)
|
||
|
|
|
||
|
|
return {
|
||
|
|
'status': 'success',
|
||
|
|
'user': user
|
||
|
|
}
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
db.rollback()
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': f'Erro na autenticação: {str(e)}'
|
||
|
|
}
|
||
|
|
finally:
|
||
|
|
db.close()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def desautenticar_usuario(user) -> Dict:
|
||
|
|
"""Desautentica um usuário"""
|
||
|
|
db = get_db_connection()
|
||
|
|
try:
|
||
|
|
if user:
|
||
|
|
user.logout()
|
||
|
|
db.commit()
|
||
|
|
logout_user()
|
||
|
|
|
||
|
|
return {
|
||
|
|
'status': 'success',
|
||
|
|
'message': 'Logout realizado com sucesso!'
|
||
|
|
}
|
||
|
|
except Exception as e:
|
||
|
|
db.rollback()
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': f'Erro no logout: {str(e)}'
|
||
|
|
}
|
||
|
|
finally:
|
||
|
|
db.close()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def alterar_senha(user_id: int, senha_atual: str, nova_senha: str) -> Dict:
|
||
|
|
"""Altera a senha de um usuário"""
|
||
|
|
db = get_db_connection()
|
||
|
|
try:
|
||
|
|
user = db.query(Usuario).get(user_id)
|
||
|
|
if not user:
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': 'Usuário não encontrado.'
|
||
|
|
}
|
||
|
|
|
||
|
|
if not user.check_password(senha_atual):
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': 'Senha atual incorreta.'
|
||
|
|
}
|
||
|
|
|
||
|
|
user.set_password(nova_senha)
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
return {
|
||
|
|
'status': 'success',
|
||
|
|
'message': 'Senha alterada com sucesso!'
|
||
|
|
}
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
db.rollback()
|
||
|
|
return {
|
||
|
|
'status': 'error',
|
||
|
|
'message': f'Erro ao alterar senha: {str(e)}'
|
||
|
|
}
|
||
|
|
finally:
|
||
|
|
db.close()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def gerar_qr_code(user) -> str:
|
||
|
|
"""Gera um QR code para o usuário"""
|
||
|
|
if not user.otp_secret:
|
||
|
|
user.otp_secret = pyotp.random_base32()
|
||
|
|
|
||
|
|
totp = pyotp.TOTP(user.otp_secret)
|
||
|
|
qr = qrcode.QRCode(version=1, box_size=10, border=5)
|
||
|
|
qr.add_data(totp.provisioning_uri(user.email, issuer_name="Sistema de Controles"))
|
||
|
|
qr.make(fit=True)
|
||
|
|
|
||
|
|
img = qr.make_image(fill_color="black", back_color="white")
|
||
|
|
buffer = BytesIO()
|
||
|
|
img.save(buffer, format="PNG")
|
||
|
|
qr_code = base64.b64encode(buffer.getvalue()).decode('utf-8')
|
||
|
|
|
||
|
|
return qr_code
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def verificar_sessao(user) -> Dict:
|
||
|
|
"""Verifica se a sessão do usuário ainda é válida"""
|
||
|
|
if not user.is_authenticated:
|
||
|
|
return {
|
||
|
|
'valid': False,
|
||
|
|
'message': 'Usuário não autenticado'
|
||
|
|
}
|
||
|
|
|
||
|
|
if user.is_session_expired():
|
||
|
|
return {
|
||
|
|
'valid': False,
|
||
|
|
'message': 'Sessão expirada'
|
||
|
|
}
|
||
|
|
|
||
|
|
return {
|
||
|
|
'valid': True
|
||
|
|
}
|