Jared Miller
6081c90ad1
Inventory saved as JSON list of thing template names in an inventory column. Migration adds column to existing databases. load_player_data returns inventory list, save_player serializes Thing names from contents.
305 lines
7.5 KiB
Python
305 lines
7.5 KiB
Python
"""SQLite persistence for player accounts and state."""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import TypedDict
|
|
|
|
from mudlib.player import Player
|
|
from mudlib.thing import Thing
|
|
|
|
|
|
class PlayerData(TypedDict):
|
|
"""Shape of persisted player data from the database."""
|
|
|
|
x: int
|
|
y: int
|
|
pl: float
|
|
stamina: float
|
|
max_stamina: float
|
|
flying: bool
|
|
zone_name: str
|
|
inventory: list[str]
|
|
|
|
|
|
# Module-level database path
|
|
_db_path: str | None = None
|
|
|
|
|
|
def init_db(db_path: str | Path) -> None:
|
|
"""Initialize the database and create tables if needed.
|
|
|
|
Args:
|
|
db_path: Path to the SQLite database file
|
|
"""
|
|
global _db_path
|
|
_db_path = str(db_path)
|
|
|
|
# Ensure parent directory exists
|
|
Path(_db_path).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
conn = sqlite3.connect(_db_path)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS accounts (
|
|
name TEXT PRIMARY KEY COLLATE NOCASE,
|
|
password_hash TEXT NOT NULL,
|
|
salt TEXT NOT NULL,
|
|
x INTEGER NOT NULL DEFAULT 0,
|
|
y INTEGER NOT NULL DEFAULT 0,
|
|
pl REAL NOT NULL DEFAULT 100.0,
|
|
stamina REAL NOT NULL DEFAULT 100.0,
|
|
max_stamina REAL NOT NULL DEFAULT 100.0,
|
|
flying INTEGER NOT NULL DEFAULT 0,
|
|
zone_name TEXT NOT NULL DEFAULT 'overworld',
|
|
inventory TEXT NOT NULL DEFAULT '[]',
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
last_login TEXT
|
|
)
|
|
""")
|
|
|
|
# Migrations: add columns if they don't exist (old schemas)
|
|
cursor.execute("PRAGMA table_info(accounts)")
|
|
columns = [row[1] for row in cursor.fetchall()]
|
|
if "zone_name" not in columns:
|
|
cursor.execute(
|
|
"ALTER TABLE accounts "
|
|
"ADD COLUMN zone_name TEXT NOT NULL DEFAULT 'overworld'"
|
|
)
|
|
if "inventory" not in columns:
|
|
cursor.execute(
|
|
"ALTER TABLE accounts ADD COLUMN inventory TEXT NOT NULL DEFAULT '[]'"
|
|
)
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def _get_connection() -> sqlite3.Connection:
|
|
"""Get a connection to the database.
|
|
|
|
Returns:
|
|
sqlite3.Connection
|
|
|
|
Raises:
|
|
RuntimeError: If init_db has not been called
|
|
"""
|
|
if _db_path is None:
|
|
raise RuntimeError("Database not initialized. Call init_db() first.")
|
|
return sqlite3.connect(_db_path)
|
|
|
|
|
|
def _hash_password(password: str, salt: bytes) -> str:
|
|
"""Hash a password with the given salt.
|
|
|
|
Args:
|
|
password: The password to hash
|
|
salt: Salt bytes
|
|
|
|
Returns:
|
|
Hex-encoded hash
|
|
"""
|
|
return hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode("utf-8"), salt, iterations=100000
|
|
).hex()
|
|
|
|
|
|
def create_account(name: str, password: str) -> bool:
|
|
"""Create a new account with the given name and password.
|
|
|
|
Args:
|
|
name: Account name (case-insensitive)
|
|
password: Password to hash and store
|
|
|
|
Returns:
|
|
True if account was created, False if name is already taken
|
|
"""
|
|
if account_exists(name):
|
|
return False
|
|
|
|
# Generate random salt
|
|
salt = os.urandom(32)
|
|
password_hash = _hash_password(password, salt)
|
|
|
|
conn = _get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
cursor.execute(
|
|
"INSERT INTO accounts (name, password_hash, salt) VALUES (?, ?, ?)",
|
|
(name, password_hash, salt.hex()),
|
|
)
|
|
conn.commit()
|
|
return True
|
|
except sqlite3.IntegrityError:
|
|
# Race condition: someone created the account between our check and insert
|
|
return False
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def account_exists(name: str) -> bool:
|
|
"""Check if an account exists.
|
|
|
|
Args:
|
|
name: Account name (case-insensitive)
|
|
|
|
Returns:
|
|
True if account exists, False otherwise
|
|
"""
|
|
conn = _get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("SELECT 1 FROM accounts WHERE name = ? LIMIT 1", (name,))
|
|
result = cursor.fetchone()
|
|
|
|
conn.close()
|
|
return result is not None
|
|
|
|
|
|
def authenticate(name: str, password: str) -> bool:
|
|
"""Verify a password for the given account.
|
|
|
|
Args:
|
|
name: Account name (case-insensitive)
|
|
password: Password to verify
|
|
|
|
Returns:
|
|
True if password is correct, False otherwise
|
|
"""
|
|
conn = _get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("SELECT password_hash, salt FROM accounts WHERE name = ?", (name,))
|
|
result = cursor.fetchone()
|
|
conn.close()
|
|
|
|
if result is None:
|
|
return False
|
|
|
|
stored_hash, salt_hex = result
|
|
salt = bytes.fromhex(salt_hex)
|
|
password_hash = _hash_password(password, salt)
|
|
|
|
return hmac.compare_digest(password_hash, stored_hash)
|
|
|
|
|
|
def save_player(player: Player) -> None:
|
|
"""Save player state to the database.
|
|
|
|
The account must already exist. This updates game state only.
|
|
|
|
Args:
|
|
player: Player instance to save
|
|
"""
|
|
# Serialize inventory as JSON list of thing names
|
|
inventory_names = [obj.name for obj in player.contents if isinstance(obj, Thing)]
|
|
inventory_json = json.dumps(inventory_names)
|
|
|
|
conn = _get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(
|
|
"""
|
|
UPDATE accounts
|
|
SET x = ?, y = ?, pl = ?, stamina = ?, max_stamina = ?, flying = ?,
|
|
zone_name = ?, inventory = ?
|
|
WHERE name = ?
|
|
""",
|
|
(
|
|
player.x,
|
|
player.y,
|
|
player.pl,
|
|
player.stamina,
|
|
player.max_stamina,
|
|
1 if player.flying else 0,
|
|
player.location.name if player.location else "overworld",
|
|
inventory_json,
|
|
player.name,
|
|
),
|
|
)
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def load_player_data(name: str) -> PlayerData | None:
|
|
"""Load player data from the database.
|
|
|
|
Args:
|
|
name: Account name (case-insensitive)
|
|
|
|
Returns:
|
|
Dictionary of persisted fields, or None if account not found
|
|
"""
|
|
conn = _get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# Check which columns exist (for migration compatibility)
|
|
cursor.execute("PRAGMA table_info(accounts)")
|
|
columns = [row[1] for row in cursor.fetchall()]
|
|
has_zone_name = "zone_name" in columns
|
|
has_inventory = "inventory" in columns
|
|
|
|
# Build SELECT based on available columns
|
|
select_cols = "x, y, pl, stamina, max_stamina, flying"
|
|
if has_zone_name:
|
|
select_cols += ", zone_name"
|
|
if has_inventory:
|
|
select_cols += ", inventory"
|
|
|
|
cursor.execute(
|
|
f"SELECT {select_cols} FROM accounts WHERE name = ?",
|
|
(name,),
|
|
)
|
|
|
|
result = cursor.fetchone()
|
|
conn.close()
|
|
|
|
if result is None:
|
|
return None
|
|
|
|
# Unpack base fields
|
|
x, y, pl, stamina, max_stamina, flying_int = result[:6]
|
|
idx = 6
|
|
|
|
zone_name = "overworld"
|
|
if has_zone_name:
|
|
zone_name = result[idx]
|
|
idx += 1
|
|
|
|
inventory: list[str] = []
|
|
if has_inventory:
|
|
inventory = json.loads(result[idx])
|
|
|
|
return {
|
|
"x": x,
|
|
"y": y,
|
|
"pl": pl,
|
|
"stamina": stamina,
|
|
"max_stamina": max_stamina,
|
|
"flying": bool(flying_int),
|
|
"zone_name": zone_name,
|
|
"inventory": inventory,
|
|
}
|
|
|
|
|
|
def update_last_login(name: str) -> None:
|
|
"""Update the last_login timestamp for an account.
|
|
|
|
Args:
|
|
name: Account name (case-insensitive)
|
|
"""
|
|
conn = _get_connection()
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute(
|
|
"UPDATE accounts SET last_login = datetime('now') WHERE name = ?", (name,)
|
|
)
|
|
|
|
conn.commit()
|
|
conn.close()
|