- padronizando o nome de get_db_connection e session para get_db_session, para não confundir com session do Flask ou sessoes web
- corrigindo potenciais erros
-- has_permission nao consegue com lazy load carregar permission depois de load_user fechar a conexao, entao joinedLoad com Permission antes de fechar
-- db.rollback não existe caso db = get_db_session() apareça muito depois dentro do try, padronizando antes de try
--- comparar role por nivel (Role.SECRETARIO_GERAL) e nao por nome ("Secretario Geral")
- unificacao de get_otp_qr_code
- mudança de nowutc() para now(UTC) conforme novo padrão
142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
from functions.database import get_db_session, Usuario
|
|
from flask_login import login_user, logout_user
|
|
from datetime import datetime
|
|
from typing import Dict, Optional
|
|
from services.otp_service import generate_qr_code_base64
|
|
|
|
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_session()
|
|
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_session()
|
|
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_session()
|
|
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"""
|
|
return generate_qr_code_base64(user)
|
|
|
|
@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
|
|
}
|