118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
import os
|
||
import sqlite3
|
||
|
||
DB_FILE = "/var/lib/bot/users.db"
|
||
|
||
def get_db_connection():
|
||
"""Obtiene una conexión a la base de datos SQLite."""
|
||
conn = sqlite3.connect(DB_FILE)
|
||
conn.row_factory = sqlite3.Row # Permitir acceder a columnas por nombre
|
||
return conn
|
||
|
||
def init_db():
|
||
"""Inicializa la base de datos de autenticación si no existe."""
|
||
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
|
||
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
|
||
# Tabla para usuarios registrados
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
nick TEXT UNIQUE NOT NULL,
|
||
password_hash TEXT NOT NULL
|
||
)
|
||
""")
|
||
|
||
# Tabla para sesiones activas
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS authenticated_users (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
nick TEXT UNIQUE NOT NULL
|
||
)
|
||
""")
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
def is_authenticated(nick):
|
||
"""Verifica si un usuario está autenticado."""
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT nick FROM authenticated_users WHERE nick = ?", (nick,))
|
||
row = cursor.fetchone()
|
||
conn.close()
|
||
return row is not None # Si el usuario está en la tabla, está autenticado
|
||
|
||
def authenticate_user(nick):
|
||
"""Autentica a un usuario agregándolo a la base de datos."""
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
cursor.execute("INSERT OR IGNORE INTO authenticated_users (nick) VALUES (?)", (nick,))
|
||
conn.commit()
|
||
conn.close()
|
||
return f" {nick} ha iniciado sesión correctamente."
|
||
|
||
def logout_user(nick):
|
||
"""Cierra la sesión de un usuario."""
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
cursor.execute("DELETE FROM authenticated_users WHERE nick = ?", (nick,))
|
||
conn.commit()
|
||
conn.close()
|
||
return f" {nick} ha cerrado sesión."
|
||
|
||
def register_user(nick, password):
|
||
"""Registra un nuevo usuario en la base de datos con una contraseña."""
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
cursor.execute("INSERT INTO users (nick, password_hash) VALUES (?, ?)", (nick, password))
|
||
conn.commit()
|
||
response = f" Usuario {nick} registrado correctamente."
|
||
except sqlite3.IntegrityError:
|
||
response = f"️ El usuario {nick} ya está registrado."
|
||
|
||
conn.close()
|
||
return response
|
||
|
||
def login_user(nick, password):
|
||
"""Autentica al usuario verificando su contraseña."""
|
||
conn = get_db_connection()
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT password_hash FROM users WHERE nick = ?", (nick,))
|
||
row = cursor.fetchone()
|
||
|
||
if row and row["password_hash"] == password:
|
||
return authenticate_user(nick)
|
||
else:
|
||
return f" Contraseña incorrecta para {nick}."
|
||
|
||
def run(sender, *args):
|
||
"""Ejecuta el comando `.auth`."""
|
||
if not args:
|
||
return " Uso: `.auth register <nick> <password>`, `.auth login <nick> <password>`, `.auth logout <nick>`"
|
||
|
||
action = args[0].lower()
|
||
|
||
if action == "register" and len(args) == 3:
|
||
return register_user(args[1], args[2])
|
||
|
||
elif action == "login" and len(args) == 3:
|
||
return login_user(args[1], args[2])
|
||
|
||
elif action == "logout" and len(args) == 2:
|
||
return logout_user(args[1])
|
||
|
||
return " Uso incorrecto. Prueba `.auth register <nick> <password>` o `.auth login <nick> <password>`."
|
||
|
||
def help():
|
||
"""Muestra la ayuda para el módulo de autenticación."""
|
||
return ("️ Comandos de autenticación:\n"
|
||
" - `.auth register <nick> <password>` → Registra un usuario.\n"
|
||
" - `.auth login <nick> <password>` → Inicia sesión.\n"
|
||
" - `.auth logout <nick>` → Cierra sesión.")
|
||
|