636 lines
34 KiB
Python
636 lines
34 KiB
Python
import sqlite3
|
|
import string
|
|
import secrets
|
|
import io
|
|
import zoneinfo
|
|
import os
|
|
import markdown
|
|
from datetime import datetime, timezone
|
|
from passlib.hash import argon2
|
|
import pyotp
|
|
import qrcode
|
|
import qrcode.image.svg
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
from apscheduler.triggers.date import DateTrigger
|
|
from flask import Flask, request, session, redirect, render_template, url_for, abort, send_from_directory
|
|
from flask_socketio import SocketIO, emit, join_room
|
|
|
|
application = Flask(__name__)
|
|
application.secret_key = secrets.token_hex(64)
|
|
websocket_manager = SocketIO(application, manage_session=False, async_mode='threading')
|
|
scheduler_instance = BackgroundScheduler(timezone="Europe/Berlin")
|
|
european_timezone = zoneinfo.ZoneInfo("Europe/Berlin")
|
|
|
|
IDENTIFIER_LENGTH = 5
|
|
|
|
def get_database_connection():
|
|
connection = sqlite3.connect("game_database.sqlite")
|
|
connection.row_factory = sqlite3.Row
|
|
return connection
|
|
|
|
def initialize_database():
|
|
connection = get_database_connection()
|
|
connection.execute("""
|
|
CREATE TABLE IF NOT EXISTS game_configuration (
|
|
configuration_key TEXT PRIMARY KEY,
|
|
configuration_value TEXT
|
|
)
|
|
""")
|
|
connection.execute("""
|
|
CREATE TABLE IF NOT EXISTS player (
|
|
player_identifier TEXT PRIMARY KEY,
|
|
password_hash TEXT,
|
|
display_name TEXT,
|
|
is_alive INTEGER DEFAULT 1,
|
|
score INTEGER DEFAULT 0,
|
|
target_identifier TEXT,
|
|
reset_token TEXT,
|
|
pending_attacker_identifier TEXT,
|
|
is_approved INTEGER DEFAULT 0,
|
|
pending_display_name TEXT,
|
|
late_joiner INTEGER DEFAULT 0
|
|
)
|
|
""")
|
|
connection.execute("""
|
|
CREATE TABLE IF NOT EXISTS administrator (
|
|
administrator_identifier INTEGER PRIMARY KEY,
|
|
password_hash TEXT,
|
|
two_factor_secret TEXT,
|
|
two_factor_enabled INTEGER DEFAULT 0
|
|
)
|
|
""")
|
|
connection.execute("""
|
|
CREATE TABLE IF NOT EXISTS kill_log (
|
|
kill_identifier INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
killer_identifier TEXT,
|
|
victim_identifier TEXT,
|
|
victim_previous_target TEXT,
|
|
kill_timestamp TEXT
|
|
)
|
|
""")
|
|
try:
|
|
connection.execute("ALTER TABLE administrator ADD COLUMN two_factor_enabled INTEGER DEFAULT 0")
|
|
except sqlite3.OperationalError:
|
|
pass
|
|
try:
|
|
connection.execute("ALTER TABLE player ADD COLUMN is_approved INTEGER DEFAULT 0")
|
|
except sqlite3.OperationalError:
|
|
pass
|
|
try:
|
|
connection.execute("ALTER TABLE player ADD COLUMN pending_display_name TEXT")
|
|
except sqlite3.OperationalError:
|
|
pass
|
|
try:
|
|
connection.execute("ALTER TABLE player ADD COLUMN late_joiner INTEGER DEFAULT 0")
|
|
except sqlite3.OperationalError:
|
|
pass
|
|
cursor = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'")
|
|
if not cursor.fetchone():
|
|
connection.execute("INSERT INTO game_configuration (configuration_key, configuration_value) VALUES ('game_status', 'pending')")
|
|
connection.execute("INSERT INTO game_configuration (configuration_key, configuration_value) VALUES ('start_time', '')")
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
def require_player_authentication(function):
|
|
def wrapper(*arguments, **keyword_arguments):
|
|
if "player_identifier" not in session:
|
|
return redirect(url_for("handle_login"))
|
|
return function(*arguments, **keyword_arguments)
|
|
wrapper.__name__ = function.__name__
|
|
return wrapper
|
|
|
|
def require_administrator_authentication(function):
|
|
def wrapper(*arguments, **keyword_arguments):
|
|
if not session.get("administrator_authenticated"):
|
|
return redirect(url_for("handle_administrator"))
|
|
return function(*arguments, **keyword_arguments)
|
|
wrapper.__name__ = function.__name__
|
|
return wrapper
|
|
|
|
def get_markdown_rules():
|
|
if os.path.exists("rules.md"):
|
|
with open("rules.md", "r", encoding="utf-8") as markdown_file:
|
|
return markdown.markdown(markdown_file.read())
|
|
return "<h2>Regelwerk nicht gefunden</h2>"
|
|
|
|
def shuffle_player_targets():
|
|
connection = get_database_connection()
|
|
players = connection.execute("SELECT player_identifier FROM player WHERE is_alive = 1 AND is_approved = 1").fetchall()
|
|
if len(players) > 1:
|
|
identifiers = [player_record["player_identifier"] for player_record in players]
|
|
secure_random_generator = secrets.SystemRandom()
|
|
secure_random_generator.shuffle(identifiers)
|
|
for index, identifier in enumerate(identifiers):
|
|
target = identifiers[(index + 1) % len(identifiers)]
|
|
connection.execute("UPDATE player SET target_identifier = ? WHERE player_identifier = ?", (target, identifier))
|
|
connection.commit()
|
|
connection.close()
|
|
websocket_manager.emit("game_state_changed")
|
|
|
|
def process_administrator_action(action_type, form_data):
|
|
connection = get_database_connection()
|
|
has_game_state_changed = False
|
|
|
|
if action_type == "start_game":
|
|
alive_count_record = connection.execute("SELECT COUNT(*) as alive_count FROM player WHERE is_approved = 1").fetchone()
|
|
if alive_count_record["alive_count"] >= 2:
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'running' WHERE configuration_key = 'game_status'")
|
|
connection.execute("UPDATE game_configuration SET configuration_value = ? WHERE configuration_key = 'start_time'", (datetime.now(european_timezone).isoformat(),))
|
|
connection.commit()
|
|
connection.close()
|
|
shuffle_player_targets()
|
|
websocket_manager.emit("admin_state_changed", to="administrator_room")
|
|
return
|
|
|
|
elif action_type == "stop_game":
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'finished' WHERE configuration_key = 'game_status'")
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "pause_game":
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'paused' WHERE configuration_key = 'game_status'")
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "resume_game":
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'running' WHERE configuration_key = 'game_status'")
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "reset_game":
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'pending' WHERE configuration_key = 'game_status'")
|
|
connection.execute("UPDATE game_configuration SET configuration_value = '' WHERE configuration_key = 'start_time'")
|
|
connection.execute("UPDATE player SET is_alive = 1, late_joiner = 0, score = 0, target_identifier = NULL, pending_attacker_identifier = NULL")
|
|
connection.execute("DELETE FROM kill_log")
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "create_player":
|
|
display_name = form_data.get("display_name")
|
|
new_identifier = ''.join(secrets.choice(string.ascii_uppercase) for _ in range(IDENTIFIER_LENGTH))
|
|
temporary_password = secrets.token_urlsafe(16)
|
|
hashed_password = argon2.hash(temporary_password)
|
|
connection.execute("INSERT INTO player (player_identifier, password_hash, display_name, is_approved) VALUES (?, ?, ?, 0)", (new_identifier, hashed_password, display_name))
|
|
connection.commit()
|
|
|
|
elif action_type == "approve_player":
|
|
target_identifier = form_data.get("player_identifier")
|
|
status_record = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'").fetchone()
|
|
is_late_joiner = 1 if status_record["configuration_value"] in ['running', 'paused'] else 0
|
|
is_alive_status = 0 if is_late_joiner else 1
|
|
connection.execute("UPDATE player SET is_approved = 1, is_alive = ?, late_joiner = ? WHERE player_identifier = ?", (is_alive_status, is_late_joiner, target_identifier))
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "edit_name":
|
|
target_identifier = form_data.get("player_identifier")
|
|
new_name = form_data.get("new_display_name")
|
|
connection.execute("UPDATE player SET display_name = ? WHERE player_identifier = ?", (new_name, target_identifier))
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "approve_name_change":
|
|
target_identifier = form_data.get("player_identifier")
|
|
connection.execute("UPDATE player SET display_name = pending_display_name, pending_display_name = NULL WHERE player_identifier = ?", (target_identifier,))
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "reject_name_change":
|
|
target_identifier = form_data.get("player_identifier")
|
|
connection.execute("UPDATE player SET pending_display_name = NULL WHERE player_identifier = ?", (target_identifier,))
|
|
connection.commit()
|
|
websocket_manager.emit("game_state_changed", to=target_identifier)
|
|
|
|
elif action_type == "delete_player":
|
|
target_identifier = form_data.get("player_identifier")
|
|
connection.execute("DELETE FROM player WHERE player_identifier = ?", (target_identifier,))
|
|
status_record = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'").fetchone()
|
|
if status_record["configuration_value"] == 'running':
|
|
alive_count_record = connection.execute("SELECT COUNT(*) as alive_count FROM player WHERE is_alive = 1 AND is_approved = 1").fetchone()
|
|
if alive_count_record["alive_count"] <= 1:
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'finished' WHERE configuration_key = 'game_status'")
|
|
if alive_count_record["alive_count"] == 1:
|
|
connection.execute("UPDATE player SET score = score + 3 WHERE is_alive = 1 AND is_approved = 1")
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "revive_player":
|
|
target_identifier = form_data.get("player_identifier")
|
|
player_to_revive = connection.execute("SELECT * FROM player WHERE player_identifier = ?", (target_identifier,)).fetchone()
|
|
if player_to_revive and player_to_revive["is_alive"] == 0:
|
|
alive_players = connection.execute("SELECT * FROM player WHERE is_alive = 1 AND is_approved = 1").fetchall()
|
|
if len(alive_players) > 0:
|
|
random_living_player = secrets.choice(alive_players)
|
|
random_living_identifier = random_living_player["player_identifier"]
|
|
attacker = connection.execute("SELECT player_identifier FROM player WHERE target_identifier = ?", (random_living_identifier,)).fetchone()
|
|
if attacker:
|
|
connection.execute("UPDATE player SET target_identifier = ? WHERE player_identifier = ?", (target_identifier, attacker["player_identifier"]))
|
|
connection.execute("UPDATE player SET is_alive = 1, late_joiner = 0, target_identifier = ? WHERE player_identifier = ?", (random_living_identifier, target_identifier))
|
|
else:
|
|
connection.execute("UPDATE player SET is_alive = 1, late_joiner = 0, target_identifier = NULL WHERE player_identifier = ?", (target_identifier,))
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "eliminate_player":
|
|
target_identifier = form_data.get("player_identifier")
|
|
victim = connection.execute("SELECT * FROM player WHERE player_identifier = ?", (target_identifier,)).fetchone()
|
|
if victim and victim["is_alive"] == 1:
|
|
attacker = connection.execute("SELECT player_identifier FROM player WHERE target_identifier = ?", (target_identifier,)).fetchone()
|
|
new_target = victim["target_identifier"]
|
|
connection.execute("UPDATE player SET is_alive = 0, target_identifier = NULL, pending_attacker_identifier = NULL WHERE player_identifier = ?", (target_identifier,))
|
|
|
|
attacker_identifier = None
|
|
if attacker:
|
|
attacker_identifier = attacker["player_identifier"]
|
|
connection.execute("UPDATE player SET target_identifier = ? WHERE player_identifier = ?", (new_target, attacker_identifier))
|
|
|
|
timestamp = datetime.now(european_timezone).isoformat()
|
|
connection.execute("INSERT INTO kill_log (killer_identifier, victim_identifier, victim_previous_target, kill_timestamp) VALUES (?, ?, ?, ?)", (attacker_identifier, target_identifier, new_target, timestamp))
|
|
|
|
alive_count_record = connection.execute("SELECT COUNT(*) as alive_count FROM player WHERE is_alive = 1 AND is_approved = 1").fetchone()
|
|
if alive_count_record["alive_count"] <= 1:
|
|
status_record = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'").fetchone()
|
|
current_status = status_record["configuration_value"] if status_record else "unknown"
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'finished' WHERE configuration_key = 'game_status'")
|
|
|
|
if alive_count_record["alive_count"] == 1 and current_status != 'finished':
|
|
connection.execute("UPDATE player SET score = score + 3 WHERE is_alive = 1 AND is_approved = 1")
|
|
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "edit_kill":
|
|
kill_identifier = form_data.get("kill_identifier")
|
|
new_killer_identifier = form_data.get("new_killer_identifier")
|
|
if new_killer_identifier == "":
|
|
new_killer_identifier = None
|
|
new_timestamp = form_data.get("kill_timestamp")
|
|
|
|
old_kill = connection.execute("SELECT * FROM kill_log WHERE kill_identifier = ?", (kill_identifier,)).fetchone()
|
|
if old_kill:
|
|
old_killer_identifier = old_kill["killer_identifier"]
|
|
if old_killer_identifier != new_killer_identifier:
|
|
if old_killer_identifier:
|
|
connection.execute("UPDATE player SET score = MAX(0, score - 1) WHERE player_identifier = ?", (old_killer_identifier,))
|
|
if new_killer_identifier:
|
|
connection.execute("UPDATE player SET score = score + 1 WHERE player_identifier = ?", (new_killer_identifier,))
|
|
|
|
connection.execute("UPDATE kill_log SET killer_identifier = ?, kill_timestamp = ? WHERE kill_identifier = ?", (new_killer_identifier, new_timestamp, kill_identifier))
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "revoke_kill":
|
|
kill_identifier = form_data.get("kill_identifier")
|
|
kill_record = connection.execute("SELECT * FROM kill_log WHERE kill_identifier = ?", (kill_identifier,)).fetchone()
|
|
|
|
if kill_record:
|
|
if kill_record["killer_identifier"]:
|
|
connection.execute("UPDATE player SET score = MAX(0, score - 1) WHERE player_identifier = ?", (kill_record["killer_identifier"],))
|
|
|
|
victim_identifier = kill_record["victim_identifier"]
|
|
target_to_restore = kill_record["victim_previous_target"]
|
|
|
|
player_pointing_to_target = connection.execute("SELECT player_identifier FROM player WHERE target_identifier = ?", (target_to_restore,)).fetchone()
|
|
|
|
connection.execute("UPDATE player SET is_alive = 1, target_identifier = ? WHERE player_identifier = ?", (target_to_restore, victim_identifier))
|
|
if player_pointing_to_target:
|
|
connection.execute("UPDATE player SET target_identifier = ? WHERE player_identifier = ?", (victim_identifier, player_pointing_to_target["player_identifier"]))
|
|
|
|
connection.execute("DELETE FROM kill_log WHERE kill_identifier = ?", (kill_identifier,))
|
|
|
|
alive_count = connection.execute("SELECT COUNT(*) as alive_count FROM player WHERE is_alive = 1 AND is_approved = 1").fetchone()["alive_count"]
|
|
if alive_count > 1:
|
|
status = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'").fetchone()["configuration_value"]
|
|
if status == 'finished':
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'running' WHERE configuration_key = 'game_status'")
|
|
connection.execute("UPDATE player SET score = MAX(0, score - 3) WHERE is_alive = 1 AND is_approved = 1 AND player_identifier != ?", (victim_identifier,))
|
|
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "change_target":
|
|
player_identifier = form_data.get("player_identifier")
|
|
new_target_identifier = form_data.get("new_target_identifier")
|
|
if player_identifier and new_target_identifier and player_identifier != new_target_identifier:
|
|
player_record = connection.execute("SELECT target_identifier FROM player WHERE player_identifier = ?", (player_identifier,)).fetchone()
|
|
other_player_record = connection.execute("SELECT player_identifier FROM player WHERE target_identifier = ?", (new_target_identifier,)).fetchone()
|
|
if player_record and other_player_record:
|
|
connection.execute("UPDATE player SET target_identifier = ? WHERE player_identifier = ?", (new_target_identifier, player_identifier))
|
|
connection.execute("UPDATE player SET target_identifier = ? WHERE player_identifier = ?", (player_record["target_identifier"], other_player_record["player_identifier"]))
|
|
connection.commit()
|
|
has_game_state_changed = True
|
|
|
|
elif action_type == "generate_reset":
|
|
target_identifier = form_data.get("player_identifier")
|
|
reset_token = secrets.token_urlsafe(32)
|
|
connection.execute("UPDATE player SET reset_token = ? WHERE player_identifier = ?", (reset_token, target_identifier))
|
|
connection.commit()
|
|
|
|
elif action_type == "update_credentials":
|
|
new_password = form_data.get("password_string")
|
|
password_confirmation = form_data.get("password_string_confirmation")
|
|
if new_password and new_password == password_confirmation:
|
|
hashed_password = argon2.hash(new_password)
|
|
connection.execute("UPDATE administrator SET password_hash = ? WHERE administrator_identifier = 1", (hashed_password,))
|
|
connection.commit()
|
|
|
|
elif action_type == "enable_two_factor":
|
|
two_factor_input = form_data.get("two_factor_code")
|
|
administrator_record = connection.execute("SELECT two_factor_secret FROM administrator WHERE administrator_identifier = 1").fetchone()
|
|
time_based_one_time_password_validator = pyotp.TOTP(administrator_record["two_factor_secret"])
|
|
if time_based_one_time_password_validator.verify(two_factor_input):
|
|
connection.execute("UPDATE administrator SET two_factor_enabled = 1 WHERE administrator_identifier = 1")
|
|
connection.commit()
|
|
|
|
elif action_type == "disable_two_factor":
|
|
connection.execute("UPDATE administrator SET two_factor_enabled = 0 WHERE administrator_identifier = 1")
|
|
connection.commit()
|
|
|
|
connection.close()
|
|
|
|
if has_game_state_changed:
|
|
websocket_manager.emit("game_state_changed")
|
|
else:
|
|
websocket_manager.emit("admin_state_changed", to="administrator_room")
|
|
|
|
def scheduled_action_wrapper(action_type, form_data):
|
|
with application.app_context():
|
|
process_administrator_action(action_type, form_data)
|
|
|
|
@application.route("/service_worker.js")
|
|
def serve_service_worker():
|
|
return send_from_directory(application.static_folder, "service_worker.js")
|
|
|
|
@application.route("/")
|
|
@require_player_authentication
|
|
def handle_index():
|
|
connection = get_database_connection()
|
|
player = connection.execute("SELECT * FROM player WHERE player_identifier = ?", (session["player_identifier"],)).fetchone()
|
|
if not player:
|
|
session.clear()
|
|
return redirect(url_for("handle_login"))
|
|
status_row = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'").fetchone()
|
|
game_status = status_row["configuration_value"]
|
|
time_row = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'start_time'").fetchone()
|
|
start_time = time_row["configuration_value"]
|
|
|
|
target_player = None
|
|
if player["target_identifier"]:
|
|
target_player = connection.execute("SELECT display_name FROM player WHERE player_identifier = ?", (player["target_identifier"],)).fetchone()
|
|
|
|
pending_attacker_name = None
|
|
if player["pending_attacker_identifier"]:
|
|
attacker = connection.execute("SELECT display_name FROM player WHERE player_identifier = ?", (player["pending_attacker_identifier"],)).fetchone()
|
|
if attacker:
|
|
pending_attacker_name = attacker["display_name"]
|
|
|
|
all_players = connection.execute("SELECT * FROM player WHERE is_approved = 1 ORDER BY score DESC").fetchall()
|
|
total_eliminations = sum(player_record["score"] for player_record in all_players)
|
|
alive_count = sum(1 for player_record in all_players if player_record["is_alive"] == 1)
|
|
connection.close()
|
|
|
|
document_markup = get_markdown_rules()
|
|
|
|
return render_template(
|
|
"game.html",
|
|
player=player,
|
|
game_status=game_status,
|
|
start_time=start_time,
|
|
target_player=target_player,
|
|
pending_attacker_name=pending_attacker_name,
|
|
all_players=all_players,
|
|
total_eliminations=total_eliminations,
|
|
alive_count=alive_count,
|
|
rules_markup=document_markup
|
|
)
|
|
|
|
@application.route("/register", methods=["GET", "POST"])
|
|
def handle_registration():
|
|
if request.method == "POST":
|
|
display_name = request.form.get("display_name")
|
|
password_input = request.form.get("password_string")
|
|
password_confirmation = request.form.get("password_string_confirmation")
|
|
if password_input and password_input == password_confirmation:
|
|
new_identifier = ''.join(secrets.choice(string.ascii_uppercase) for _ in range(IDENTIFIER_LENGTH))
|
|
hashed_password = argon2.hash(password_input)
|
|
connection = get_database_connection()
|
|
connection.execute("INSERT INTO player (player_identifier, password_hash, display_name, is_approved) VALUES (?, ?, ?, 0)", (new_identifier, hashed_password, display_name))
|
|
connection.commit()
|
|
connection.close()
|
|
websocket_manager.emit("admin_state_changed", to="administrator_room")
|
|
return render_template("login.html", identifier_length=IDENTIFIER_LENGTH, registration_success=True, new_identifier=new_identifier)
|
|
return render_template("login.html", identifier_length=IDENTIFIER_LENGTH, register_mode=True)
|
|
|
|
@application.route("/login", methods=["GET", "POST"])
|
|
def handle_login():
|
|
if request.method == "POST":
|
|
identifier_input = request.form.get("player_identifier", "").strip().upper()
|
|
password_input = request.form.get("password_string", "")
|
|
connection = get_database_connection()
|
|
player = connection.execute("SELECT password_hash FROM player WHERE player_identifier = ?", (identifier_input,)).fetchone()
|
|
connection.close()
|
|
if player and argon2.verify(password_input, player["password_hash"]):
|
|
session["player_identifier"] = identifier_input
|
|
return redirect(url_for("handle_index"))
|
|
return render_template("login.html", identifier_length=IDENTIFIER_LENGTH)
|
|
|
|
@application.route("/reset_password", methods=["GET", "POST"])
|
|
def handle_password_reset():
|
|
token = request.args.get("token") or request.form.get("token")
|
|
if not token:
|
|
abort(400)
|
|
connection = get_database_connection()
|
|
player = connection.execute("SELECT player_identifier FROM player WHERE reset_token = ?", (token,)).fetchone()
|
|
if not player:
|
|
connection.close()
|
|
abort(403)
|
|
if request.method == "POST":
|
|
new_password = request.form.get("password_string")
|
|
password_confirmation = request.form.get("password_string_confirmation")
|
|
if new_password and new_password == password_confirmation:
|
|
hashed_password = argon2.hash(new_password)
|
|
connection.execute("UPDATE player SET password_hash = ?, reset_token = NULL WHERE reset_token = ?", (hashed_password, token))
|
|
connection.commit()
|
|
connection.close()
|
|
return redirect(url_for("handle_login"))
|
|
connection.close()
|
|
return render_template("login.html", identifier_length=IDENTIFIER_LENGTH, reset_mode=True, token=token)
|
|
|
|
@application.route("/player/action", methods=["POST"])
|
|
@require_player_authentication
|
|
def handle_player_action():
|
|
action_type = request.form.get("action_type")
|
|
connection = get_database_connection()
|
|
if action_type == "request_name_change":
|
|
pending_name = request.form.get("pending_display_name")
|
|
connection.execute("UPDATE player SET pending_display_name = ? WHERE player_identifier = ?", (pending_name, session["player_identifier"]))
|
|
connection.commit()
|
|
websocket_manager.emit("admin_state_changed", to="administrator_room")
|
|
connection.close()
|
|
return redirect(url_for("handle_index"))
|
|
|
|
@application.route("/logout")
|
|
def handle_logout():
|
|
session.clear()
|
|
return redirect(url_for("handle_login"))
|
|
|
|
@application.route("/admin", methods=["GET", "POST"])
|
|
def handle_administrator():
|
|
connection = get_database_connection()
|
|
administrator_record = connection.execute("SELECT * FROM administrator WHERE administrator_identifier = 1").fetchone()
|
|
|
|
if not administrator_record:
|
|
if request.method == "POST":
|
|
password_input = request.form.get("password_string")
|
|
password_confirmation = request.form.get("password_string_confirmation")
|
|
if password_input and password_input == password_confirmation:
|
|
hashed_password = argon2.hash(password_input)
|
|
two_factor_secret = pyotp.random_base32()
|
|
connection.execute("INSERT INTO administrator (administrator_identifier, password_hash, two_factor_secret, two_factor_enabled) VALUES (1, ?, ?, 0)", (hashed_password, two_factor_secret))
|
|
connection.commit()
|
|
session["administrator_authenticated"] = True
|
|
connection.close()
|
|
return redirect(url_for("handle_administrator"))
|
|
connection.close()
|
|
return render_template("administrator.html", setup_required=True)
|
|
|
|
if not session.get("administrator_authenticated"):
|
|
if request.method == "POST":
|
|
password_input = request.form.get("password_string")
|
|
if argon2.verify(password_input, administrator_record["password_hash"]):
|
|
if administrator_record["two_factor_enabled"]:
|
|
two_factor_input = request.form.get("two_factor_code")
|
|
time_based_one_time_password_validator = pyotp.TOTP(administrator_record["two_factor_secret"])
|
|
if time_based_one_time_password_validator.verify(two_factor_input):
|
|
session["administrator_authenticated"] = True
|
|
connection.close()
|
|
return redirect(url_for("handle_administrator"))
|
|
else:
|
|
session["administrator_authenticated"] = True
|
|
connection.close()
|
|
return redirect(url_for("handle_administrator"))
|
|
connection.close()
|
|
return render_template("administrator.html", login_required=True, two_factor_enabled=bool(administrator_record["two_factor_enabled"]))
|
|
|
|
status_row = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'").fetchone()
|
|
game_status = status_row["configuration_value"]
|
|
all_players = connection.execute("SELECT * FROM player ORDER BY is_approved ASC, display_name ASC").fetchall()
|
|
alive_count_record = connection.execute("SELECT COUNT(*) as alive_count FROM player WHERE is_approved = 1").fetchone()
|
|
can_start = alive_count_record["alive_count"] >= 2
|
|
|
|
kill_logs = connection.execute("""
|
|
SELECT k.*,
|
|
killer.display_name as killer_name,
|
|
victim.display_name as victim_name
|
|
FROM kill_log k
|
|
LEFT JOIN player killer ON k.killer_identifier = killer.player_identifier
|
|
LEFT JOIN player victim ON k.victim_identifier = victim.player_identifier
|
|
ORDER BY k.kill_timestamp DESC
|
|
""").fetchall()
|
|
|
|
two_factor_uri = pyotp.totp.TOTP(administrator_record["two_factor_secret"]).provisioning_uri(name="Administrator", issuer_name="AssassinGame")
|
|
qr_factory = qrcode.image.svg.SvgPathImage
|
|
qr_image = qrcode.make(two_factor_uri, image_factory=qr_factory)
|
|
qr_stream = io.BytesIO()
|
|
qr_image.save(qr_stream)
|
|
qr_code_svg = qr_stream.getvalue().decode("utf-8")
|
|
|
|
connection.close()
|
|
|
|
return render_template(
|
|
"administrator.html",
|
|
dashboard_active=True,
|
|
game_status=game_status,
|
|
can_start=can_start,
|
|
all_players=all_players,
|
|
kill_logs=kill_logs,
|
|
two_factor_secret=administrator_record["two_factor_secret"],
|
|
two_factor_enabled=bool(administrator_record["two_factor_enabled"]),
|
|
qr_code_svg=qr_code_svg
|
|
)
|
|
|
|
@application.route("/admin/action", methods=["POST"])
|
|
@require_administrator_authentication
|
|
def handle_administrator_action():
|
|
form_data = request.form.to_dict()
|
|
action_type = form_data.get("action_type")
|
|
execution_time_string = form_data.get("execution_time")
|
|
|
|
if execution_time_string:
|
|
execution_datetime = datetime.strptime(execution_time_string[:16], "%Y-%m-%dT%H:%M").replace(tzinfo=european_timezone)
|
|
scheduler_instance.add_job(scheduled_action_wrapper, DateTrigger(run_date=execution_datetime), args=[action_type, form_data])
|
|
else:
|
|
process_administrator_action(action_type, form_data)
|
|
|
|
return redirect(url_for("handle_administrator"))
|
|
|
|
@websocket_manager.on("connect")
|
|
def handle_websocket_connection():
|
|
if "player_identifier" in session:
|
|
join_room(session["player_identifier"])
|
|
if session.get("administrator_authenticated"):
|
|
join_room("administrator_room")
|
|
|
|
@websocket_manager.on("request_elimination")
|
|
def handle_elimination_request():
|
|
attacker_identifier = session.get("player_identifier")
|
|
if not attacker_identifier:
|
|
return
|
|
connection = get_database_connection()
|
|
attacker = connection.execute("SELECT target_identifier FROM player WHERE player_identifier = ?", (attacker_identifier,)).fetchone()
|
|
if attacker and attacker["target_identifier"]:
|
|
victim_identifier = attacker["target_identifier"]
|
|
connection.execute("UPDATE player SET pending_attacker_identifier = ? WHERE player_identifier = ?", (attacker_identifier, victim_identifier))
|
|
connection.commit()
|
|
emit("elimination_verification_requested", to=victim_identifier)
|
|
connection.close()
|
|
|
|
@websocket_manager.on("confirm_elimination")
|
|
def handle_elimination_confirmation():
|
|
victim_identifier = session.get("player_identifier")
|
|
if not victim_identifier:
|
|
return
|
|
connection = get_database_connection()
|
|
victim = connection.execute("SELECT pending_attacker_identifier, target_identifier FROM player WHERE player_identifier = ?", (victim_identifier,)).fetchone()
|
|
if victim and victim["pending_attacker_identifier"]:
|
|
attacker_identifier = victim["pending_attacker_identifier"]
|
|
new_target = victim["target_identifier"]
|
|
|
|
connection.execute("UPDATE player SET is_alive = 0, target_identifier = NULL, pending_attacker_identifier = NULL WHERE player_identifier = ?", (victim_identifier,))
|
|
connection.execute("UPDATE player SET target_identifier = ?, score = score + 1 WHERE player_identifier = ?", (new_target, attacker_identifier))
|
|
|
|
timestamp = datetime.now(european_timezone).isoformat()
|
|
connection.execute("INSERT INTO kill_log (killer_identifier, victim_identifier, victim_previous_target, kill_timestamp) VALUES (?, ?, ?, ?)", (attacker_identifier, victim_identifier, new_target, timestamp))
|
|
|
|
alive_count_record = connection.execute("SELECT COUNT(*) as alive_count FROM player WHERE is_alive = 1 AND is_approved = 1").fetchone()
|
|
if alive_count_record["alive_count"] <= 1:
|
|
status_record = connection.execute("SELECT configuration_value FROM game_configuration WHERE configuration_key = 'game_status'").fetchone()
|
|
current_status = status_record["configuration_value"] if status_record else "unknown"
|
|
connection.execute("UPDATE game_configuration SET configuration_value = 'finished' WHERE configuration_key = 'game_status'")
|
|
|
|
if alive_count_record["alive_count"] == 1 and current_status != 'finished':
|
|
connection.execute("UPDATE player SET score = score + 3 WHERE is_alive = 1 AND is_approved = 1")
|
|
|
|
connection.commit()
|
|
websocket_manager.emit("game_state_changed")
|
|
connection.close()
|
|
|
|
@websocket_manager.on("reject_elimination")
|
|
def handle_elimination_rejection():
|
|
victim_identifier = session.get("player_identifier")
|
|
if not victim_identifier:
|
|
return
|
|
connection = get_database_connection()
|
|
victim = connection.execute("SELECT pending_attacker_identifier FROM player WHERE player_identifier = ?", (victim_identifier,)).fetchone()
|
|
if victim and victim["pending_attacker_identifier"]:
|
|
attacker_identifier = victim["pending_attacker_identifier"]
|
|
connection.execute("UPDATE player SET pending_attacker_identifier = NULL WHERE player_identifier = ?", (victim_identifier,))
|
|
connection.commit()
|
|
emit("elimination_rejected", to=attacker_identifier)
|
|
connection.close()
|
|
|
|
if __name__ == "__main__":
|
|
initialize_database()
|
|
scheduler_instance.start()
|
|
websocket_manager.run(application, host="0.0.0.0", port=5000) |