Код: Выделить всё
authenticating...
2024-05-10 01:41:58,350 - INFO - authenticating locally...
2024-05-10 01:42:03,663 - ERROR - An unexpected error occurred: database is locked
2024-05-10 01:42:03,663 - INFO - Database connection closed.
2024-05-10 01:42:03,663 - INFO - User admin authenticated successfully.
Я не знаю, почему он это делает, я все перепробовал Я думал об этом, но это не работает.
если вы можете мне помочь с этим, я буду благодарен.
код:
Authenticator.py:
Код: Выделить всё
import sqlite3, secrets, logging
# to provide password hashing utilites
from passlib.hash import argon2
from datahandler import checkInternet
from AddToken import add_token_to_database
# Configures the logging system to display all logs with a severity level of INFO or higher.
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
# Creates a logger with the name ‘authenticator’for logging messages within this module.
logger = logging.getLogger(__name__)
class DatabaseManager:
def __init__(self, db_path):
# Establishes a connection to the SQLite database at the specified path.
self.conn = sqlite3.connect(db_path)
# Creates a cursor object to execute SQL commands.
self.cursor = self.conn.cursor()
def close(self):
# Closes the SQLite database connection.
self.conn.close()
class Authenticator:
def __init__(self, db_manager):
self.db_manager = db_manager
async def authenticate_online(self, username, password):
# Placeholder for online authentication logic
return False, None
def authenticate_locally(self, username, password):
logger.info("authenticating locally...")
try:
self.db_manager.cursor.execute(
"SELECT Password FROM Users WHERE Username = ?", (username,)
)
result = self.db_manager.cursor.fetchone()
if result is None:
logger.error(f"No user found with the username: {username}")
return False, None
hashed_password = result[0]
if hashed_password and argon2.verify(password, hashed_password):
token = create_session_token()
add_token_to_database(username, token)
logger.info(f"User {username} authenticated successfully.")
return True, token
else:
logger.error(f"Password verification failed for user: {username}")
return False, None
except sqlite3.Error as e:
logger.error(f"SQLite error during local authentication: {e}")
return False, None
except ValueError as e:
logger.error(f"Malformed Argon2 hash, or other library error: {e}")
return False, None
async def authenticate(self, username, password):
print("authenticating...")
if checkInternet("www.google.com", 3):
# online authentication
success, token = await self.authenticate_online(username, password)
else:
# offline authentication
success, token = self.authenticate_locally(username, password)
if success:
# Ensures that a valid token is returned, either a new one or an existing valid session token.
token = token or self.get_valid_session_token(username)
return success, token
# retrieves a valid session token for the user if it exists.
def get_valid_session_token(self, username):
self.db_manager.cursor.execute(
"SELECT Token FROM Sessions WHERE Username = ?", (username,)
)
result = self.db_manager.cursor.fetchone()
if result:
return result[0]
return None
# function to generate a secure session token using the secrets module
def create_session_token():
return secrets.token_urlsafe(16)
Код: Выделить всё
import sqlite3, logging
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def add_token_to_database(username, token, db_path="DB.db"):
try:
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"""
INSERT OR REPLACE INTO Sessions (Username, Token) VALUES (?, ?)
""",
(username, token),
)
conn.commit()
logger.info("Session token inserted/updated successfully for user: %s", username)
except sqlite3.IntegrityError as e:
logger.error(f"Integrity error when adding token: {e}")
except Exception as e:
logger.error(f"An unexpected error occurred: {e}")
finally:
logger.info("Database connection closed.")
Код: Выделить всё
import json, logging
from Authenticator import DatabaseManager, Authenticator
db_path = "./DB.db"
db_manager = DatabaseManager(db_path)
authenticator = Authenticator(db_manager)
# Initialize logger
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# An asynchronous function that listens for login requests over a WebSocket connection
# and It decodes the JSON message, extracts the username and password, and attempts to authenticate the user
# If authentication is successful, it sends back a JSON response with a success flag and a session token.
async def handle_login(websocket, path):
async for message in websocket:
try:
data = json.loads(message)
if data.get("action") == "login":
username = data["username"]
password = data["password"]
authenticated, token = await authenticator.authenticate(
username, password
)
if authenticated:
await websocket.send(json.dumps({"success": True, "token": token}))
else:
await websocket.send(json.dumps({"success": False}))
except json.JSONDecodeError:
await websocket.send(
json.dumps({"success": False, "error": "Invalid JSON"})
)
except Exception as e:
logger.error("Login handling error: %s", e)
await websocket.send(
json.dumps({"success": False, "error": "Internal server error"})
)
Подробнее здесь: https://stackoverflow.com/questions/784 ... the-databa
