commit 65abbe4aa6331f795fb482f32339a16aef979372 Author: AAsige <168331480+AAsige@users.noreply.github.com> Date: Thu Jul 30 20:56:50 2026 +0800 Add files via upload diff --git a/linux-native-backend-v1-usable/Dockerfile b/linux-native-backend-v1-usable/Dockerfile new file mode 100644 index 0000000..d6a3028 --- /dev/null +++ b/linux-native-backend-v1-usable/Dockerfile @@ -0,0 +1,15 @@ +FROM python:3.14.6-slim-bookworm + +ENV PYTHONDONTWRITEBYTECODE=1 +ENV PYTHONUNBUFFERED=1 +ENV HOST=0.0.0.0 +ENV PORT=8765 + +WORKDIR /app + +COPY app ./app +COPY data ./data + +EXPOSE 8765 + +CMD ["python", "-m", "app.server", "--host", "0.0.0.0", "--port", "8765"] diff --git a/linux-native-backend-v1-usable/README.md b/linux-native-backend-v1-usable/README.md new file mode 100644 index 0000000..b0b51f8 --- /dev/null +++ b/linux-native-backend-v1-usable/README.md @@ -0,0 +1,157 @@ +# Brass Birmingham Linux Native Backend + +这是基于现有客户端协议重写的 Linux 原生后端。 + +## 当前状态 + +这份实现不是从源码直接移植，而是基于下面两部分反推出来的： + +- `BrassHost.exe` 中提取出的 Python 服务端字节码 +- `Assembly-CSharp.dll` 中提取出的客户端接口字符串 + +它已经覆盖了客户端联机所需的核心接口： + +- 登录 +- 注册 +- 当前用户 +- 房间列表 +- 创建房间 +- 加入房间 +- 离开房间 +- 事件同步 +- `lastSeenRevision` 更新 + +## 目录结构 + +```text +linux-native-backend/ + app/ + __init__.py + server.py + data/ + games.json + users.json + run.sh + test_local.py +``` + +## Docker 启动 + +先构建 image： + +```bash +cd linux-native-backend +chmod +x build-image.sh +./build-image.sh +``` + +默认会生成： + +```text +brass-birmingham-backend:latest +``` + +如果你想自定义名字或 tag： + +```bash +IMAGE_NAME=myrepo/brass-backend IMAGE_TAG=v1 ./build-image.sh +``` + +推荐直接用 Docker Compose： + +```bash +cd linux-native-backend +docker compose up -d --build +``` + +查看日志： + +```bash +docker compose logs -f +``` + +停止服务： + +```bash +docker compose down +``` + +当前 `Dockerfile` 使用的是明确存在的基础镜像标签： + +```text +python:3.14.6-slim-bookworm +``` + +这样比浮动别名更稳，也更方便排查镜像拉取问题。 + +## 目录挂载 + +`docker-compose.yml` 已经把本地数据目录挂载到容器里： + +```text +./data -> /app/data +``` + +所以容器重建后，用户和房间数据仍会保留。 + +## 手动构建与运行 + +如果你不想用 Compose，也可以直接执行： + +```bash +cd linux-native-backend +docker build -t brass-birmingham-backend . +docker run -d \ + --name brass-birmingham-backend \ + -p 8765:8765 \ + -v "$(pwd)/data:/app/data" \ + --restart unless-stopped \ + brass-birmingham-backend +``` + +## 直接本机启动 + +```bash +cd linux-native-backend +chmod +x run.sh +./run.sh +``` + +默认监听： + +```text +0.0.0.0:8765 +``` + +## API 概览 + +- `POST /login` +- `POST /login/with/e-mail` +- `POST /register` +- `GET /me` +- `POST /logout` +- `POST /password/email` +- `GET /1.0.0/games` +- `POST /1.0.0/games` +- `PUT /1.0.0/games//players` +- `DELETE /1.0.0/games//players` +- `GET /1.0.0/games//events` +- `POST /1.0.0/games//events` +- `DELETE /1.0.0/games//events` +- `GET /1.0.0/games//events-and-metadata` +- `POST /1.0.0/games//metadata/lastSeenRevision` +- `POST /1.0.0/devices` +- `GET /debug/games` + +## 认证方式 + +后端兼容原工具里暴露出来的伪 token 机制： + +- `Authorization: Bearer fake-token-` +- 或请求头 `X-User-Id: ` + +## 已知限制 + +- 这份协议实现是逆向恢复版，不保证 100% 覆盖所有隐藏边缘逻辑 +- 目前没有实现 ngrok 辅助启动，因为这不属于后端核心协议 +- 如果后续抓到真实客户端请求样本，还可以继续把返回结构再对齐得更严 diff --git a/linux-native-backend-v1-usable/VERSION_SNAPSHOT.md b/linux-native-backend-v1-usable/VERSION_SNAPSHOT.md new file mode 100644 index 0000000..4d94b6f --- /dev/null +++ b/linux-native-backend-v1-usable/VERSION_SNAPSHOT.md @@ -0,0 +1,13 @@ +# First Usable Version + +This directory is the current working version of the Linux native backend. + +Snapshot label: + +`v1-usable` + +Notes: + +- Captures the current Docker-based backend layout. +- Includes the current protocol compatibility work for login, room listing, join flow, and turn progression. +- Intended as the first rollback point before further fixes. diff --git a/linux-native-backend-v1-usable/__pycache__/test_turn_progression.cpython-312.pyc b/linux-native-backend-v1-usable/__pycache__/test_turn_progression.cpython-312.pyc new file mode 100644 index 0000000..0a69423 Binary files /dev/null and b/linux-native-backend-v1-usable/__pycache__/test_turn_progression.cpython-312.pyc differ diff --git a/linux-native-backend-v1-usable/__pycache__/test_turn_progression.cpython-314.pyc b/linux-native-backend-v1-usable/__pycache__/test_turn_progression.cpython-314.pyc new file mode 100644 index 0000000..828970a Binary files /dev/null and b/linux-native-backend-v1-usable/__pycache__/test_turn_progression.cpython-314.pyc differ diff --git a/linux-native-backend-v1-usable/app/__init__.py b/linux-native-backend-v1-usable/app/__init__.py new file mode 100644 index 0000000..8603468 --- /dev/null +++ b/linux-native-backend-v1-usable/app/__init__.py @@ -0,0 +1 @@ +"""Brass Birmingham Linux-native backend.""" diff --git a/linux-native-backend-v1-usable/app/__pycache__/__init__.cpython-312.pyc b/linux-native-backend-v1-usable/app/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..273a4f5 Binary files /dev/null and b/linux-native-backend-v1-usable/app/__pycache__/__init__.cpython-312.pyc differ diff --git a/linux-native-backend-v1-usable/app/__pycache__/__init__.cpython-314.pyc b/linux-native-backend-v1-usable/app/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..32cb069 Binary files /dev/null and b/linux-native-backend-v1-usable/app/__pycache__/__init__.cpython-314.pyc differ diff --git a/linux-native-backend-v1-usable/app/__pycache__/server.cpython-312.pyc b/linux-native-backend-v1-usable/app/__pycache__/server.cpython-312.pyc new file mode 100644 index 0000000..ad2127f Binary files /dev/null and b/linux-native-backend-v1-usable/app/__pycache__/server.cpython-312.pyc differ diff --git a/linux-native-backend-v1-usable/app/__pycache__/server.cpython-314.pyc b/linux-native-backend-v1-usable/app/__pycache__/server.cpython-314.pyc new file mode 100644 index 0000000..e13ba30 Binary files /dev/null and b/linux-native-backend-v1-usable/app/__pycache__/server.cpython-314.pyc differ diff --git a/linux-native-backend-v1-usable/app/server.py b/linux-native-backend-v1-usable/app/server.py new file mode 100644 index 0000000..12d6b4f --- /dev/null +++ b/linux-native-backend-v1-usable/app/server.py @@ -0,0 +1,1332 @@ +from __future__ import annotations + +import argparse +import json +import logging +import re +import socket +import threading +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, urlparse + + +ROOT_DIR = Path(__file__).resolve().parent.parent +DATA_DIR = ROOT_DIR / "data" +USERS_FILE = DATA_DIR / "users.json" +GAMES_FILE = DATA_DIR / "games.json" +SERVER_PORT = 8765 +LOGGER = logging.getLogger("brass_backend") + + +def now_iso() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def get_local_ip() -> str: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.connect(("8.8.8.8", 80)) + return sock.getsockname()[0] + except OSError: + return "127.0.0.1" + finally: + sock.close() + + +def load_json(path: Path, default: Any) -> Any: + if not path.exists(): + return default + with path.open("r", encoding="utf-8") as handle: + return json.load(handle) + + +def save_json(path: Path, payload: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False, indent=2) + + +def ensure_player_metadata_slots(game: dict[str, Any]) -> None: + cfg = game.setdefault("GameStartConfiguration", {}) + wanted = int(cfg.get("NumberOfPlayersAsInt", 4) or 4) + players = game.setdefault("PlayerMetadata", []) + while len(players) < wanted: + players.append({}) + + +def normalize_revision(value: Any) -> int: + try: + return int(value) + except (TypeError, ValueError): + return 0 + + +def make_response(data: Any = None, status: int = 200) -> tuple[int, dict[str, Any]]: + return status, {"data": data} + + +def make_error(message: str, status: int = 400) -> tuple[int, dict[str, Any]]: + return status, {"error": message} + + +def make_auth_response(access_token: str, status: int = 200) -> tuple[int, dict[str, Any]]: + return status, {"access_token": access_token} + + +def make_user_response(user: dict[str, Any], status: int = 200) -> tuple[int, dict[str, Any]]: + return status, { + "id": user["id"], + "email": user["email"], + "name": user.get("name") or user.get("nickname") or user["email"], + } + + +def make_message_response(message: str = "ok", status: int = 200) -> tuple[int, dict[str, Any]]: + return status, {"message": message} + + +def coerce_payload_dict(body: Any) -> dict[str, Any]: + if not isinstance(body, dict): + return {} + nested = body.get("data") + if isinstance(nested, dict): + merged = dict(nested) + for key, value in body.items(): + if key != "data" and key not in merged: + merged[key] = value + return merged + return body + + +def nested_get(payload: Any, path: list[str]) -> Any: + current = payload + for part in path: + if isinstance(current, dict): + current = current.get(part) + else: + return None + return current + + +def make_raw_response(payload: Any, status: int = 200) -> tuple[int, Any]: + return status, payload + + +def ensure_game_metadata_shape(game: dict[str, Any]) -> dict[str, Any]: + metadata = game.setdefault("metadata", {}) + if "PlayerMetadata" not in metadata and "PlayerMetadata" in game: + metadata["PlayerMetadata"] = game["PlayerMetadata"] + if "NetworkGameProgressInformation" not in metadata and "NetworkGameProgressInformation" in game: + metadata["NetworkGameProgressInformation"] = game["NetworkGameProgressInformation"] + if "GameStartConfiguration" not in metadata and "GameStartConfiguration" in game: + metadata["GameStartConfiguration"] = game["GameStartConfiguration"] + if "events" not in metadata and "events" in game: + metadata["events"] = game["events"] + return game + + +def infer_nickname_from_game_name(game_name: Any) -> str | None: + if not isinstance(game_name, str) or not game_name: + return None + if "'的游戏" in game_name: + return game_name.split("'的游戏", 1)[0].strip() or None + if "'s game" in game_name: + return game_name.split("'s game", 1)[0].strip() or None + return None + + +def metadata_values(metadata: dict[str, Any], key: str) -> list[Any]: + value = metadata.get(key) + if isinstance(value, list): + return value + if isinstance(value, dict) and isinstance(value.get("$values"), list): + return value["$values"] + return [] + + +def normalize_player_entry(entry: Any) -> Any: + if entry is None or not isinstance(entry, dict): + return entry + if isinstance(entry.get("metadata"), dict): + normalized = dict(entry["metadata"]) + for key in ( + "PlayerNetworkId", + "NickName", + "GameId", + "AiType", + "PlayerAvatar", + "PlayerStartConfiguration", + "LastSeenEventRevision", + ): + if entry.get(key) is not None and normalized.get(key) is None: + normalized[key] = entry.get(key) + return normalized + return entry + + +def normalize_player_metadata_slots(metadata: dict[str, Any]) -> list[Any]: + players = metadata.get("PlayerMetadata", []) + if isinstance(players, dict) and "$values" in players: + players = players["$values"] + metadata["PlayerMetadata"] = players + if not isinstance(players, list): + players = [] + metadata["PlayerMetadata"] = players + + for idx, entry in enumerate(players): + players[idx] = normalize_player_entry(entry) + + max_players = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) + try: + max_players = int(max_players) + except (TypeError, ValueError): + max_players = len(players) + + while len(players) < max_players: + players.append(None) + if max_players > 0 and len(players) > max_players: + del players[max_players:] + return players + + +def real_player_count(players: list[Any]) -> int: + return sum(1 for entry in players if isinstance(entry, dict)) + + +def sync_network_game_status(metadata: dict[str, Any]) -> list[Any]: + players = normalize_player_metadata_slots(metadata) + max_players = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) + try: + max_players = int(max_players) + except (TypeError, ValueError): + max_players = len(players) + + progress = metadata.setdefault("NetworkGameProgressInformation", {}) + joined = real_player_count(players) + if joined >= max_players and max_players > 0: + progress["CurrentGameStatus"] = "InProgress" + if not progress.get("CurrentPlayerId"): + first_player = next((entry for entry in players if isinstance(entry, dict)), None) + if first_player: + progress["CurrentPlayerId"] = first_player.get("PlayerNetworkId") + elif joined > 0: + progress["CurrentGameStatus"] = "Open" + return players + + +def event_revision(event: dict[str, Any]) -> int: + if not isinstance(event, dict): + return 0 + nested = event.get("Metadata") + if isinstance(nested, dict): + try: + return int(nested.get("Revision") or 0) + except (TypeError, ValueError): + return 0 + try: + return int(event.get("Revision") or 0) + except (TypeError, ValueError): + return 0 + + +def set_event_revision(event: dict[str, Any], revision: int) -> None: + metadata = event.setdefault("Metadata", {}) + if isinstance(metadata, dict): + metadata["Revision"] = revision + event["Revision"] = revision + + +def normalize_events_payload(events: Any) -> list[Any]: + if isinstance(events, dict) and isinstance(events.get("$values"), list): + return list(events["$values"]) + if isinstance(events, list): + flattened: list[Any] = [] + for event in events: + if isinstance(event, dict) and isinstance(event.get("$values"), list): + flattened.extend(event["$values"]) + else: + flattened.append(event) + return flattened + return [events] + + +def _resolve_player_id_from_color(metadata: dict[str, Any], player_color: Any) -> str | None: + if player_color is None: + return None + + players = metadata_values(metadata, "PlayerMetadata") + for player in players: + if not isinstance(player, dict): + continue + + start_cfg = player.get("PlayerStartConfiguration") + if isinstance(start_cfg, dict) and start_cfg.get("Color") == player_color: + return player.get("PlayerNetworkId") or None + + if isinstance(player.get("PlayerStartConfiguration"), dict): + cfg_color = player.get("PlayerStartConfiguration", {}).get("Color") + if cfg_color == player_color: + return player.get("PlayerNetworkId") or None + + if player.get("NickName") and str(player.get("NickName")).lower() == str(player_color).lower(): + return player.get("PlayerNetworkId") or None + + return None + + +def _resolve_next_player_id(metadata: dict[str, Any], current_player_color: Any) -> str | None: + if current_player_color is None: + return None + + start_configs = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) + if not isinstance(start_configs, list): + return None + + ordered_colors = [] + for entry in start_configs: + if isinstance(entry, dict): + color = entry.get("Color") + if color is not None: + ordered_colors.append(str(color)) + + if not ordered_colors: + return None + + try: + current_index = ordered_colors.index(str(current_player_color)) + except ValueError: + return None + + next_index = (current_index + 1) % len(ordered_colors) + next_color = ordered_colors[next_index] + return _resolve_player_id_from_color(metadata, next_color) + + +def _resolve_next_player(metadata: dict[str, Any], current_player_color: Any) -> tuple[str | None, str | None]: + if current_player_color is None: + return None, None + + start_configs = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) + if not isinstance(start_configs, list): + return None, None + + ordered_colors = [] + for entry in start_configs: + if isinstance(entry, dict): + color = entry.get("Color") + if color is not None: + ordered_colors.append(str(color)) + + if not ordered_colors: + return None, None + + try: + current_index = ordered_colors.index(str(current_player_color)) + except ValueError: + return None, None + + next_index = (current_index + 1) % len(ordered_colors) + next_color = ordered_colors[next_index] + next_player_id = _resolve_player_id_from_color(metadata, next_color) + return next_player_id, next_color + + +def sync_progress_from_events(game: dict[str, Any]) -> None: + metadata = game.setdefault("metadata", {}) + progress = metadata.setdefault("NetworkGameProgressInformation", {}) + game_progress = metadata.setdefault("GameProgressInformation", {}) + events = game.get("events", []) + + if not isinstance(events, list): + return + + current_player_id = None + current_player_color = None + + for event in reversed(events): + if not isinstance(event, dict): + continue + + event_type = str(event.get("$type") or "") + + if "TurnStarted" in event_type and event.get("PlayerColor") is not None: + current_player_id = _resolve_player_id_from_color(metadata, event.get("PlayerColor")) + current_player_color = str(event.get("PlayerColor")) + break + + if "TurnEnded" in event_type: + previous_turn_color = None + for candidate in reversed(events[: events.index(event)]): + if not isinstance(candidate, dict): + continue + candidate_type = str(candidate.get("$type") or "") + if "TurnStarted" in candidate_type and candidate.get("PlayerColor") is not None: + previous_turn_color = str(candidate.get("PlayerColor")) + break + if previous_turn_color is not None: + current_player_id, current_player_color = _resolve_next_player(metadata, previous_turn_color) + break + + if current_player_id: + progress["CurrentPlayerId"] = current_player_id + progress["CurrentGameStatus"] = "InProgress" + progress["CurrentPlayerColor"] = current_player_color or progress.get("CurrentPlayerColor") + current_player = game_progress.setdefault("CurrentPlayer", {}) + if current_player_color: + current_player["Color"] = current_player_color + return + + for event in reversed(events): + if not isinstance(event, dict): + continue + event_type = str(event.get("$type") or "") + if "TurnStarted" in event_type and event.get("PlayerColor") is not None: + player_color = event.get("PlayerColor") + player_id = _resolve_player_id_from_color(metadata, player_color) + if player_id: + progress["CurrentPlayerId"] = player_id + progress["CurrentGameStatus"] = "InProgress" + progress["CurrentPlayerColor"] = str(player_color) + current_player = game_progress.setdefault("CurrentPlayer", {}) + current_player["Color"] = str(player_color) + break + + +def game_display_name(game: dict[str, Any]) -> str: + metadata = game.get("metadata", {}) + return str(nested_get(metadata, ["NetworkGameStartConfiguration", "GameName"]) or "") + + +def game_owner_key(game: dict[str, Any]) -> str: + metadata = game.get("metadata", {}) + players = normalize_player_metadata_slots(metadata) + first_player = next((entry for entry in players if isinstance(entry, dict)), None) + if first_player: + return str(first_player.get("PlayerNetworkId") or first_player.get("NickName") or "") + nickname = infer_nickname_from_game_name(game_display_name(game)) + return nickname or "" + + +def game_dedupe_key(game: dict[str, Any]) -> str: + return f"{game_owner_key(game)}|{game_display_name(game)}" + + +def is_client_safe_game_metadata(metadata: Any) -> bool: + if not isinstance(metadata, dict): + return False + if not metadata.get("$type"): + return False + if not isinstance(metadata.get("GameStartConfiguration"), dict): + return False + if not isinstance(metadata.get("NetworkGameProgressInformation"), dict): + return False + if not isinstance(metadata.get("NetworkGameStartConfiguration"), dict): + return False + player_metadata = metadata.get("PlayerMetadata") + if isinstance(player_metadata, dict): + if not isinstance(player_metadata.get("$values"), list): + return False + elif not isinstance(player_metadata, list): + return False + return True + + +def cleanup_games_on_load(raw_games: list[Any], users: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: + games_by_id: dict[str, dict[str, Any]] = {} + user_names = { + user.get("id", ""): user.get("name") or user.get("nickname") or user.get("email") or "" + for user in users + if isinstance(user, dict) + } + + deduped: dict[str, dict[str, Any]] = {} + removed_ids: list[str] = [] + + for entry in raw_games: + if not isinstance(entry, dict) or "id" not in entry: + continue + game = ensure_game_metadata_shape(dict(entry)) + metadata = game.setdefault("metadata", {}) + + # Normalize PlayerMetadata container shape. + normalize_player_metadata_slots(metadata) + + # Repair empty or anonymous owner slot from game name when possible. + players = metadata.get("PlayerMetadata", []) + real_players = [p for p in players if isinstance(p, dict)] + if not real_players: + nickname = infer_nickname_from_game_name(game_display_name(game)) + matched_user_id = None + for uid, uname in user_names.items(): + if uname == nickname: + matched_user_id = uid + break + start_players = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) + first_start = start_players[0] if isinstance(start_players, list) and start_players else None + inferred = { + "$type": "BrassApplication.Game.PlayerMetadata, Assembly-CSharp", + "NickName": nickname or "Host", + "AiType": "Human", + "PlayerAvatar": "Arkwright", + "PlayerStartConfiguration": first_start, + "LastSeenEventRevision": 0, + } + if matched_user_id: + inferred["PlayerNetworkId"] = matched_user_id + player_count = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) + try: + player_count = int(player_count) + except (TypeError, ValueError): + player_count = 2 + players = [inferred] + while len(players) < max(player_count, 1): + players.append(None) + metadata["PlayerMetadata"] = players + + sync_network_game_status(metadata) + game["updated_at"] = game.get("updated_at") or game.get("created_at") or now_iso() + games_by_id[game["id"]] = game + + for game in games_by_id.values(): + key = game_dedupe_key(game) or game.get("id", "") + existing = deduped.get(key) + if existing is None: + deduped[key] = game + continue + existing_players = real_player_count(normalize_player_metadata_slots(existing.get("metadata", {}))) + current_players = real_player_count(normalize_player_metadata_slots(game.get("metadata", {}))) + existing_updated = str(existing.get("updated_at") or "") + current_updated = str(game.get("updated_at") or "") + if current_players > existing_players or ( + current_players == existing_players and current_updated >= existing_updated + ): + removed_ids.append(existing["id"]) + deduped[key] = game + else: + removed_ids.append(game["id"]) + + if removed_ids: + LOGGER.warning("[BOOT] removed duplicate/stale game ids=%s", removed_ids) + + return {game["id"]: game for game in deduped.values()} + + +def setup_logging() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s [%(name)s] %(message)s", + ) + + +@dataclass +class Storage: + lock: threading.Lock + users: list[dict[str, Any]] + games: dict[str, dict[str, Any]] + tokens: dict[str, str] + + @classmethod + def load(cls) -> "Storage": + DATA_DIR.mkdir(parents=True, exist_ok=True) + users = load_json(USERS_FILE, []) + games_list = load_json(GAMES_FILE, []) + games = cleanup_games_on_load(games_list, users) + return cls(lock=threading.Lock(), users=users, games=games, tokens={}) + + def save_users(self) -> None: + save_json(USERS_FILE, self.users) + + def save_games(self) -> None: + save_json(GAMES_FILE, list(self.games.values())) + + def find_user_by_email(self, email: str) -> dict[str, Any] | None: + email = email.strip().lower() + for user in self.users: + if user.get("email", "").strip().lower() == email: + return user + return None + + def user_by_id(self, user_id: str) -> dict[str, Any] | None: + for user in self.users: + if user.get("id") == user_id: + return user + return None + + +class BrassAPI: + def __init__(self, storage: Storage) -> None: + self.storage = storage + + def repair_game_metadata_for_listing(self, game: dict[str, Any]) -> dict[str, Any]: + game = ensure_game_metadata_shape(dict(game)) + metadata = game.setdefault("metadata", {}) + players = metadata.get("PlayerMetadata") + if isinstance(players, dict) and "$values" in players: + players = players["$values"] + metadata["PlayerMetadata"] = players + + if isinstance(players, list) and len(players) == 0: + game_name = nested_get(metadata, ["NetworkGameStartConfiguration", "GameName"]) + nickname = infer_nickname_from_game_name(game_name) + matched_user = None + if nickname: + for user in self.storage.users: + if user.get("name") == nickname or user.get("nickname") == nickname: + matched_user = user + break + + start_players = nested_get(metadata, ["GameStartConfiguration", "Players", "$values"]) + first_start = start_players[0] if isinstance(start_players, list) and start_players else None + inferred_player = { + "$type": "BrassApplication.Game.PlayerMetadata, Assembly-CSharp", + "NickName": nickname or "Host", + "AiType": "Human", + "PlayerAvatar": "Arkwright", + "PlayerStartConfiguration": first_start, + "LastSeenEventRevision": 0, + } + if matched_user: + inferred_player["PlayerNetworkId"] = matched_user["id"] + + player_count = nested_get(metadata, ["GameStartConfiguration", "NumberOfPlayersAsInt"]) + try: + player_count = int(player_count) + except (TypeError, ValueError): + player_count = 2 + + repaired_players: list[Any] = [inferred_player] + while len(repaired_players) < max(player_count, 1): + repaired_players.append(None) + + metadata["PlayerMetadata"] = repaired_players + LOGGER.warning( + "[GAME] repaired empty PlayerMetadata game_id=%s nickname=%s inferred_user_id=%s", + game.get("id"), + nickname, + matched_user.get("id") if matched_user else None, + ) + + return game + + def parse_body(self, handler: BaseHTTPRequestHandler) -> Any: + length = int(handler.headers.get("Content-Length", "0") or "0") + raw = handler.rfile.read(length) if length else b"" + if not raw: + return {} + + ctype = handler.headers.get("Content-Type", "") + if "application/json" in ctype: + return json.loads(raw.decode("utf-8")) + if "application/x-www-form-urlencoded" in ctype: + form = parse_qs(raw.decode("utf-8"), keep_blank_values=True) + return {key: values[-1] if values else "" for key, values in form.items()} + try: + return json.loads(raw.decode("utf-8")) + except json.JSONDecodeError: + form = parse_qs(raw.decode("utf-8"), keep_blank_values=True) + return {key: values[-1] if values else "" for key, values in form.items()} + + def get_param(self, handler: BaseHTTPRequestHandler, body: Any, key: str, default: Any = None) -> Any: + if isinstance(body, dict) and key in body: + return body[key] + query = parse_qs(urlparse(handler.path).query) + if key in query and query[key]: + return query[key][-1] + return default + + def current_user_id(self, handler: BaseHTTPRequestHandler) -> str | None: + auth = handler.headers.get("Authorization", "") + if auth.startswith("Bearer "): + token = auth.removeprefix("Bearer ").strip() + if token in self.storage.tokens: + user_id = self.storage.tokens[token] + LOGGER.info("[AUTH] bearer token resolved user_id=%s path=%s", user_id, handler.path) + return user_id + if token.startswith("fake-token-"): + user_id = token.removeprefix("fake-token-").strip() + LOGGER.info("[AUTH] fake token resolved user_id=%s path=%s", user_id, handler.path) + return user_id + token = handler.headers.get("access_token", "") + if token: + if token in self.storage.tokens: + user_id = self.storage.tokens[token] + LOGGER.info("[AUTH] access_token resolved user_id=%s path=%s", user_id, handler.path) + return user_id + if token.startswith("fake-token-"): + user_id = token.removeprefix("fake-token-").strip() + LOGGER.info("[AUTH] fake access_token resolved user_id=%s path=%s", user_id, handler.path) + return user_id + direct = handler.headers.get("X-User-Id", "") + if direct: + user_id = direct.strip() + LOGGER.info("[AUTH] X-User-Id resolved user_id=%s path=%s", user_id, handler.path) + return user_id + LOGGER.warning("[AUTH] no auth headers path=%s remote=%s", handler.path, handler.client_address[0]) + return None + + def current_user(self, handler: BaseHTTPRequestHandler) -> dict[str, Any] | None: + user_id = self.current_user_id(handler) + if not user_id: + return None + user = self.storage.user_by_id(user_id) + if user: + LOGGER.info("[AUTH] user resolved user_id=%s email=%s", user_id, user.get("email")) + else: + LOGGER.warning("[AUTH] user not found for user_id=%s path=%s", user_id, handler.path) + return user + + def route(self, handler: BaseHTTPRequestHandler) -> tuple[int, dict[str, Any]]: + parsed = urlparse(handler.path) + path = parsed.path + method = handler.command.upper() + body = self.parse_body(handler) if method in {"POST", "PUT", "PATCH", "DELETE"} else {} + LOGGER.info("[REQ] %s %s remote=%s", method, path, handler.client_address[0]) + + if method == "POST" and path in {"/login", "/login/with/e-mail"}: + return self.login(body) + if method == "POST" and path == "/register": + return self.register(body) + if method == "GET" and path == "/me": + return self.me(handler) + if method == "POST" and path == "/logout": + return make_message_response("ok") + if method == "POST" and path == "/password/email": + return make_message_response("ok") + if method == "GET" and path == "/1.0.0/games": + return self.get_games(handler) + if method == "POST" and path == "/1.0.0/games": + return self.create_game(handler, body) + if method == "POST" and path == "/1.0.0/devices": + return make_response({"message": "ok"}) + if method == "GET" and path == "/debug/games": + return make_response(list(self.storage.games.values())) + + player_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/players", path) + if player_match: + game_id = player_match.group(1) + if method == "PUT": + return self.join_game(handler, game_id, body) + if method == "DELETE": + return self.leave_game(handler, game_id, body) + + events_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/events", path) + if events_match: + game_id = events_match.group(1) + if method == "GET": + return self.get_events(game_id) + if method == "POST": + return self.save_events(game_id, body) + if method == "DELETE": + return self.delete_events(game_id) + + metadata_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/metadata/lastSeenRevision", path) + if metadata_match and method == "POST": + return self.update_last_seen(handler, metadata_match.group(1), body) + + events_meta_match = re.fullmatch(r"/1\.0\.0/games/([^/]+)/events-and-metadata", path) + if events_meta_match: + if method in {"POST", "PUT"}: + return self.save_events_and_metadata(events_meta_match.group(1), body) + if method == "GET": + return self.get_game_snapshot(events_meta_match.group(1)) + + return make_error("Not found", status=404) + + def login(self, body: Any) -> tuple[int, dict[str, Any]]: + email = str(body.get("email", "")).strip().lower() if isinstance(body, dict) else "" + password = str(body.get("password", "")).strip() if isinstance(body, dict) else "" + LOGGER.info("[LOGIN] attempt email=%s", email or "") + if not email or not password: + LOGGER.warning("[LOGIN] rejected missing credentials email=%s", email or "") + return make_error("Missing email or password", status=400) + + user = self.storage.find_user_by_email(email) + if not user: + LOGGER.warning("[LOGIN] user not found email=%s", email) + return make_error("User not found", status=404) + + token = str(uuid.uuid4()) + self.storage.tokens[token] = user["id"] + LOGGER.info("[LOGIN] success email=%s user_id=%s token=%s", email, user["id"], token) + return 200, {"access_token": token, "data": {"access_token": token, "user": make_user_response(user)[1]}} + + def register(self, body: Any) -> tuple[int, dict[str, Any]]: + email = str(body.get("email", "")).strip().lower() if isinstance(body, dict) else "" + password = str(body.get("password", "")).strip() if isinstance(body, dict) else "" + name = str(body.get("name", "")).strip() if isinstance(body, dict) else "" + LOGGER.info("[REGISTER] attempt email=%s name=%s", email or "", name or "") + if not email or not password: + LOGGER.warning("[REGISTER] rejected missing credentials email=%s", email or "") + return make_error("Missing email or password", status=400) + if not name: + LOGGER.warning("[REGISTER] rejected missing name email=%s", email or "") + return make_error("Missing name", status=400) + if self.storage.find_user_by_email(email): + LOGGER.warning("[REGISTER] duplicate email=%s", email) + return make_error("User already exists", status=409) + + user = { + "id": str(uuid.uuid4()), + "email": email, + "nickname": name, + "name": name, + "password": password, + "createdAt": now_iso(), + } + with self.storage.lock: + self.storage.users.append(user) + self.storage.save_users() + LOGGER.info("[REGISTER] success email=%s user_id=%s", email, user["id"]) + return 201, {"data": {"id": user["id"], "email": user["email"], "name": user["name"]}, "id": user["id"], "email": user["email"], "name": user["name"]} + + def me(self, handler: BaseHTTPRequestHandler) -> tuple[int, dict[str, Any]]: + user = self.current_user(handler) + if not user: + LOGGER.warning("[ME] unauthorized path=%s", handler.path) + return make_error("Unauthorized", status=401) + LOGGER.info("[ME] success user_id=%s email=%s", user["id"], user["email"]) + payload = {"id": user["id"], "email": user["email"], "name": user.get("name") or user.get("nickname") or user["email"]} + return 200, {"data": payload, **payload} + + def matches_filter(self, game: dict[str, Any], filters: list[str]) -> bool: + for expression in filters: + if ":" not in expression: + continue + parts = expression.split(":") + if len(parts) < 3: + continue + + field_expr = parts[0] + op = parts[1] + value = ":".join(parts[2:]) + + field_path = field_expr.split("->") + base = game + if field_path[0] == "metadata": + base = game.get("metadata", {}) + field_path = field_path[1:] + + actual = nested_get(base, field_path) if len(field_path) > 1 else base.get(field_path[0]) + actual_str = json.dumps(actual, ensure_ascii=False) if isinstance(actual, (dict, list)) else str(actual) + + if field_expr == "id" and op == "eq" and game.get("id") != value: + return False + if op == "ct" and value not in actual_str: + return False + if op == "nct" and value in actual_str: + return False + if op == "eq" and actual_str != value: + return False + if op == "ne" and actual_str == value: + return False + if op in {"gt", "gte", "lt", "lte"}: + actual_cmp = actual_str + value_cmp = value + if op == "gt" and not (actual_cmp > value_cmp): + return False + if op == "gte" and not (actual_cmp >= value_cmp): + return False + if op == "lt" and not (actual_cmp < value_cmp): + return False + if op == "lte" and not (actual_cmp <= value_cmp): + return False + return True + + def get_games(self, handler: BaseHTTPRequestHandler) -> tuple[int, dict[str, Any]]: + query = parse_qs(urlparse(handler.path).query) + filters = query.get("filter[]", []) + per_page = int(query.get("perPage", ["200"])[-1] or "200") + sort_items = query.get("sort[]", []) + + games = [self.repair_game_metadata_for_listing(game) for game in self.storage.games.values()] + filtered = [game for game in games if self.matches_filter(game, filters)] + for sort_expr in reversed(sort_items): + sort_parts = sort_expr.split(":") + if len(sort_parts) < 2: + continue + field_expr, direction = sort_parts[0], sort_parts[1].upper() + field_path = field_expr.split("->") + reverse = direction in {"DSC", "DESC"} + + def sort_value(game: dict[str, Any]) -> str: + base = game + local_path = list(field_path) + if local_path and local_path[0] == "metadata": + base = game.get("metadata", {}) + local_path = local_path[1:] + if not local_path: + return "" + if len(local_path) > 1: + return str(nested_get(base, local_path) or "") + return str(base.get(local_path[0], "")) + + filtered.sort( + key=sort_value, + reverse=reverse, + ) + + deduped: dict[str, dict[str, Any]] = {} + for game in filtered: + key = game_dedupe_key(game) or game.get("id", "") + existing = deduped.get(key) + if existing is None: + deduped[key] = game + continue + + existing_players = real_player_count(normalize_player_metadata_slots(existing.get("metadata", {}))) + current_players = real_player_count(normalize_player_metadata_slots(game.get("metadata", {}))) + existing_updated = str(existing.get("updated_at") or "") + current_updated = str(game.get("updated_at") or "") + + if current_players > existing_players or ( + current_players == existing_players and current_updated >= existing_updated + ): + deduped[key] = game + + deduped_games = list(deduped.values()) + result = deduped_games if per_page == 0 else deduped_games[:per_page] + response_items = [] + skipped_ids = [] + for game in result: + metadata = game.get("metadata", {}) + if not is_client_safe_game_metadata(metadata): + skipped_ids.append(game.get("id")) + continue + response_items.append( + { + "id": game["id"], + "metadata": metadata, + "created_at": game.get("created_at"), + "updated_at": game.get("updated_at"), + } + ) + LOGGER.info( + "[GAME] list filters=%s sort=%s per_page=%s matched=%s returned=%s", + filters, + sort_items, + per_page, + len(deduped_games), + len(response_items), + ) + if skipped_ids: + LOGGER.warning("[GAME] list skipped malformed game ids=%s", skipped_ids) + LOGGER.info( + "[GAME] list response preview=%s", + json.dumps(response_items[:2], ensure_ascii=False, default=str)[:2000], + ) + return make_response(response_items) + + def create_game(self, handler: BaseHTTPRequestHandler, body: Any) -> tuple[int, dict[str, Any]]: + user = self.current_user(handler) + if not user: + LOGGER.warning("[GAME] create unauthorized") + return make_error("Unauthorized", status=401) + payload = coerce_payload_dict(body) + if not payload: + LOGGER.warning("[GAME] create invalid payload user_id=%s", user["id"]) + return make_error("Invalid payload") + + LOGGER.info( + "[GAME] create payload user_id=%s keys=%s body=%s", + user["id"], + sorted(payload.keys()), + json.dumps(payload, ensure_ascii=False, default=str)[:1000], + ) + + game_id = str( + payload.get("GameId") + or payload.get("gameId") + or payload.get("id") + or payload.get("Id") + or "" + ).strip() + if not game_id: + game_id = str(uuid.uuid4()) + LOGGER.info("[GAME] create auto-generated game_id=%s user_id=%s", game_id, user["id"]) + + with self.storage.lock: + if game_id in self.storage.games: + LOGGER.warning("[GAME] create duplicate game_id=%s user_id=%s", game_id, user["id"]) + return make_error("Game already exists", status=409) + + cfg = payload.get("GameStartConfiguration") or payload.get("gameStartConfiguration") or {} + num_players = int( + cfg.get("NumberOfPlayersAsInt") + or cfg.get("numberOfPlayersAsInt") + or payload.get("NumberOfPlayersAsInt") + or payload.get("numberOfPlayersAsInt") + or payload.get("playersCount") + or 4 + ) + + player_entry = { + "PlayerNetworkId": user["id"], + "email": user["email"], + "name": user.get("name") or user.get("nickname"), + "nickname": user.get("nickname") or user.get("name"), + "createdAt": now_iso(), + "LastSeenEventRevision": 0, + } + + metadata = payload.get("metadata", payload) + if "PlayerMetadata" not in metadata: + metadata["PlayerMetadata"] = [player_entry] + if "NetworkGameProgressInformation" not in metadata: + metadata["NetworkGameProgressInformation"] = {"CurrentGameStatus": "Open"} + if "GameStartConfiguration" not in metadata: + metadata["GameStartConfiguration"] = { + "IsPrivate": bool(cfg.get("IsPrivate", cfg.get("isPrivate", False))), + "NumberOfPlayersAsInt": num_players, + } + sync_network_game_status(metadata) + + game = { + "id": game_id, + "metadata": metadata, + "events": [], + "created_at": now_iso(), + "updated_at": now_iso(), + } + + new_key = game_dedupe_key(game) + to_remove = [] + for existing_id, existing_game in self.storage.games.items(): + if existing_id == game_id: + continue + if game_dedupe_key(existing_game) == new_key: + to_remove.append(existing_id) + for existing_id in to_remove: + LOGGER.warning("[GAME] create removing stale duplicate game_id=%s", existing_id) + self.storage.games.pop(existing_id, None) + + self.storage.games[game_id] = game + self.storage.save_games() + LOGGER.info("[GAME] created game_id=%s owner_user_id=%s", game_id, user["id"]) + return 201, { + "id": game_id, + "data": { + "id": game_id, + "metadata": game.get("metadata", {}), + "created_at": game.get("created_at"), + "updated_at": game.get("updated_at"), + }, + } + + def join_game(self, handler: BaseHTTPRequestHandler, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: + user = self.current_user(handler) + if not user: + LOGGER.warning("[GAME] join unauthorized game_id=%s", game_id) + return make_error("Unauthorized", status=401) + + with self.storage.lock: + game = self.storage.games.get(game_id) + if not game: + LOGGER.warning("[GAME] join missing game game_id=%s user_id=%s", game_id, user["id"]) + return make_error("Game not found", status=404) + + metadata = game.setdefault("metadata", {}) + players = metadata.setdefault("PlayerMetadata", []) + for existing in players: + if isinstance(existing, dict) and existing.get("PlayerNetworkId") == user["id"]: + LOGGER.info("[GAME] join skipped already joined game_id=%s user_id=%s", game_id, user["id"]) + return 200, { + "id": game_id, + "data": { + "id": game_id, + "metadata": game.get("metadata", {}), + "created_at": game.get("created_at"), + "updated_at": game.get("updated_at"), + }, + } + + payload = coerce_payload_dict(body) + join_payload = payload.get("metadata", payload) if payload else {} + if not join_payload.get("PlayerNetworkId"): + join_payload["PlayerNetworkId"] = user["id"] + if not join_payload.get("NickName"): + join_payload["NickName"] = user.get("name") or user.get("nickname") or user["email"] + if not join_payload.get("GameId"): + join_payload["GameId"] = game_id + if not join_payload.get("LastSeenEventRevision"): + join_payload["LastSeenEventRevision"] = 0 + + slot_index = -1 + for idx, existing in enumerate(players): + if existing is None: + slot_index = idx + break + if slot_index >= 0: + players[slot_index] = join_payload + else: + players.append(join_payload) + + sync_network_game_status(metadata) + game["updated_at"] = now_iso() + self.storage.save_games() + LOGGER.info( + "[GAME] joined game_id=%s user_id=%s status=%s", + game_id, + user["id"], + nested_get(metadata, ["NetworkGameProgressInformation", "CurrentGameStatus"]), + ) + return 200, { + "id": game_id, + "data": { + "id": game_id, + "metadata": game.get("metadata", {}), + "created_at": game.get("created_at"), + "updated_at": game.get("updated_at"), + }, + } + + def leave_game(self, handler: BaseHTTPRequestHandler, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: + user = self.current_user(handler) + if not user: + LOGGER.warning("[GAME] leave unauthorized game_id=%s", game_id) + return make_error("Unauthorized", status=401) + + with self.storage.lock: + game = self.storage.games.get(game_id) + if not game: + LOGGER.warning("[GAME] leave missing game game_id=%s user_id=%s", game_id, user["id"]) + return make_error("Game not found", status=404) + + metadata = game.setdefault("metadata", {}) + players = metadata.get("PlayerMetadata", []) + if isinstance(players, dict) and "$values" in players: + players = players["$values"] + metadata["PlayerMetadata"] = players + + new_players = [] + for entry in players: + if not isinstance(entry, dict): + continue + if entry.get("PlayerNetworkId") == user["id"]: + continue + new_players.append(entry) + + metadata["PlayerMetadata"] = new_players + sync_network_game_status(metadata) + game["updated_at"] = now_iso() + self.storage.save_games() + LOGGER.info("[GAME] left game_id=%s user_id=%s", game_id, user["id"]) + return make_response(True) + + def get_events(self, game_id: str) -> tuple[int, dict[str, Any]]: + game = self.storage.games.get(game_id) + if not game: + return make_error("Game not found", status=404) + return make_response(game.get("events", [])) + + def get_game_snapshot(self, game_id: str) -> tuple[int, dict[str, Any]]: + game = self.storage.games.get(game_id) + if not game: + return make_error("Game not found", status=404) + return make_response( + { + "id": game_id, + "metadata": game.get("metadata", {}), + "events": game.get("events", []), + "created_at": game.get("created_at"), + "updated_at": game.get("updated_at"), + } + ) + + def save_events_and_metadata(self, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: + game = self.storage.games.setdefault( + game_id, + {"id": game_id, "metadata": {}, "events": [], "created_at": now_iso(), "updated_at": now_iso()}, + ) + if isinstance(body, dict): + if "events" in body: + incoming_events = normalize_events_payload(body["events"]) + max_revision = max((event_revision(evt) for evt in game.get("events", [])), default=0) + for item in incoming_events: + if not isinstance(item, dict): + continue + event = dict(item) + if not event_revision(event): + max_revision += 1 + set_event_revision(event, max_revision) + else: + max_revision = max(max_revision, event_revision(event)) + game["events"].append(event) + if "metadata" in body and isinstance(body["metadata"], dict): + game["metadata"] = body["metadata"] + sync_network_game_status(game["metadata"]) + sync_progress_from_events(game) + game["updated_at"] = now_iso() + self.storage.save_games() + LOGGER.info( + "[EVENTS_META] saved game_id=%s event_count=%s current_player_id=%s status=%s", + game_id, + len(game.get("events", [])), + nested_get(game.get("metadata", {}), ["NetworkGameProgressInformation", "CurrentPlayerId"]), + nested_get(game.get("metadata", {}), ["NetworkGameProgressInformation", "CurrentGameStatus"]), + ) + return make_response(True) + + def save_events(self, game_id: str, body: Any) -> tuple[int, dict[str, Any]]: + if not isinstance(body, dict): + return make_error("Invalid payload") + game = self.storage.games.get(game_id) + if not game: + return make_error("Game not found", status=404) + + incoming = normalize_events_payload(body.get("events", body.get("data", body))) + if not isinstance(incoming, list): + return make_error("events must be a list") + + with self.storage.lock: + events = game.setdefault("events", []) + max_revision = max((event_revision(evt) for evt in events), default=0) + for item in incoming: + if not isinstance(item, dict): + continue + event = dict(item) + if not event_revision(event): + max_revision += 1 + set_event_revision(event, max_revision) + else: + max_revision = max(max_revision, event_revision(event)) + events.append(event) + + if event.get("$type") == "GameStarted": + game.setdefault("metadata", {}).setdefault( + "NetworkGameProgressInformation", {} + )["CurrentGameStatus"] = "Started" + + if "metadata" in body and isinstance(body["metadata"], dict): + game["metadata"] = body["metadata"] + sync_network_game_status(game["metadata"]) + + sync_progress_from_events(game) + game["updated_at"] = now_iso() + ensure_game_metadata_shape(game) + self.storage.save_games() + LOGGER.info( + "[EVENTS] saved game_id=%s count=%s max_revision=%s current_player_id=%s", + game_id, + len(incoming), + max_revision, + nested_get(game.get("metadata", {}), ["NetworkGameProgressInformation", "CurrentPlayerId"]), + ) + return make_response({"saved": len(incoming), "maxRevision": max_revision}) + + def delete_events(self, game_id: str) -> tuple[int, dict[str, Any]]: + with self.storage.lock: + game = self.storage.games.get(game_id) + if not game: + return make_error("Game not found", status=404) + game["events"] = [] + game["updated_at"] = now_iso() + self.storage.save_games() + return make_message_response("ok") + + def update_last_seen( + self, handler: BaseHTTPRequestHandler, game_id: str, body: Any + ) -> tuple[int, dict[str, Any]]: + user = self.current_user(handler) + if not user: + return make_error("Unauthorized", status=401) + + game = self.storage.games.get(game_id) + if not game: + return make_error("Game not found", status=404) + + revision = normalize_revision( + body.get("LastSeenEventRevision") + if isinstance(body, dict) + else 0 + ) + + with self.storage.lock: + metadata = game.setdefault("metadata", {}) + players = metadata.setdefault("PlayerMetadata", []) + for player in players: + if not isinstance(player, dict): + continue + if player.get("PlayerNetworkId") == user["id"]: + player["LastSeenEventRevision"] = revision + game["updated_at"] = now_iso() + self.storage.save_games() + return make_response(True) + + return make_error("Player not found in game", status=404) + + +class BrassRequestHandler(BaseHTTPRequestHandler): + api: BrassAPI + + def _send(self, status: int, payload: Any) -> None: + data = json.dumps(payload, ensure_ascii=False).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self) -> None: # noqa: N802 + self.handle_request() + + def do_POST(self) -> None: # noqa: N802 + self.handle_request() + + def do_PUT(self) -> None: # noqa: N802 + self.handle_request() + + def do_DELETE(self) -> None: # noqa: N802 + self.handle_request() + + def log_message(self, format: str, *args: Any) -> None: + return + + def handle_request(self) -> None: + try: + status, payload = self.api.route(self) + except Exception as exc: # noqa: BLE001 + LOGGER.exception("[REQ] unhandled error method=%s path=%s", self.command, self.path) + status, payload = make_error(f"Internal server error: {exc}", 500) + LOGGER.info("[RES] %s %s status=%s remote=%s", self.command, self.path, status, self.client_address[0]) + self._send(status, payload) + + +def build_server(host: str, port: int) -> ThreadingHTTPServer: + storage = Storage.load() + api = BrassAPI(storage) + + class Handler(BrassRequestHandler): + pass + + Handler.api = api + server = ThreadingHTTPServer((host, port), Handler) + return server + + +def main() -> int: + setup_logging() + parser = argparse.ArgumentParser() + parser.add_argument("--host", default="0.0.0.0") + parser.add_argument("--port", type=int, default=SERVER_PORT) + args = parser.parse_args() + + storage = Storage.load() + print("=" * 60) + print(" Brass Birmingham Server") + print(f"Loaded: {len(storage.users)} users, {len(storage.games)} games") + print(f"Local: http://{get_local_ip()}:{args.port}") + print("=" * 60) + + server = build_server(args.host, args.port) + server.serve_forever() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/linux-native-backend-v1-usable/build-image.sh b/linux-native-backend-v1-usable/build-image.sh new file mode 100644 index 0000000..73b7108 --- /dev/null +++ b/linux-native-backend-v1-usable/build-image.sh @@ -0,0 +1,13 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +IMAGE_NAME="${IMAGE_NAME:-brass-birmingham-backend}" +IMAGE_TAG="${IMAGE_TAG:-latest}" + +cd "$SCRIPT_DIR" + +docker build -t "${IMAGE_NAME}:${IMAGE_TAG}" . + +echo +echo "Built image: ${IMAGE_NAME}:${IMAGE_TAG}" diff --git a/linux-native-backend-v1-usable/data/games.json b/linux-native-backend-v1-usable/data/games.json new file mode 100644 index 0000000..c713599 --- /dev/null +++ b/linux-native-backend-v1-usable/data/games.json @@ -0,0 +1,146 @@ +[ + { + "id": "test-game-001", + "GameId": "test-game-001", + "created_at": "2026-07-28T15:53:26Z", + "updated_at": "2026-07-29T20:04:38Z", + "events": [ + { + "$type": "GameStarted", + "Revision": 1 + }, + { + "$type": "GameStarted", + "Revision": 2 + }, + { + "$type": "GameStarted", + "Revision": 3 + }, + { + "$type": "GameStarted", + "Revision": 4 + }, + { + "$type": "GameStarted", + "Revision": 5 + }, + { + "$type": "GameStarted", + "Revision": 6 + }, + { + "$type": "GameStarted", + "Revision": 7 + }, + { + "$type": "GameStarted", + "Revision": 8 + }, + { + "$type": "GameStarted", + "Revision": 9 + }, + { + "$type": "GameStarted", + "Revision": 10 + }, + { + "$type": "GameStarted", + "Revision": 11 + }, + { + "$type": "GameStarted", + "Revision": 12 + }, + { + "$type": "GameStarted", + "Revision": 13 + }, + { + "$type": "GameStarted", + "Metadata": { + "Revision": 14 + }, + "Revision": 14 + }, + { + "$type": "GameStarted", + "Metadata": { + "Revision": 15 + }, + "Revision": 15 + } + ], + "PlayerMetadata": [ + { + "PlayerNetworkId": "1225a3f1-f5d1-4bb8-8d0b-1c10f8a753ba", + "email": "host@example.com", + "name": "Host", + "nickname": "Host", + "createdAt": "2026-07-28T15:53:26Z", + "LastSeenEventRevision": 0 + } + ], + "NetworkGameProgressInformation": { + "CurrentGameStatus": "Started" + }, + "GameStartConfiguration": { + "IsPrivate": false, + "NumberOfPlayersAsInt": 4 + }, + "metadata": { + "PlayerMetadata": [ + { + "PlayerNetworkId": "1225a3f1-f5d1-4bb8-8d0b-1c10f8a753ba", + "email": "host@example.com", + "name": "Host", + "nickname": "Host", + "createdAt": "2026-07-28T15:53:26Z", + "LastSeenEventRevision": 0 + }, + null, + null, + null + ], + "NetworkGameProgressInformation": { + "CurrentGameStatus": "Started" + }, + "GameStartConfiguration": { + "IsPrivate": false, + "NumberOfPlayersAsInt": 4 + }, + "events": [ + { + "$type": "GameStarted", + "Revision": 1 + }, + { + "$type": "GameStarted", + "Revision": 2 + }, + { + "$type": "GameStarted", + "Revision": 3 + }, + { + "$type": "GameStarted", + "Revision": 4 + }, + { + "$type": "GameStarted", + "Revision": 5 + }, + { + "$type": "GameStarted", + "Revision": 6 + }, + { + "$type": "GameStarted", + "Revision": 7 + } + ], + "GameProgressInformation": {} + } + } +] \ No newline at end of file diff --git a/linux-native-backend-v1-usable/data/users.json b/linux-native-backend-v1-usable/data/users.json new file mode 100644 index 0000000..7904372 --- /dev/null +++ b/linux-native-backend-v1-usable/data/users.json @@ -0,0 +1,10 @@ +[ + { + "id": "1225a3f1-f5d1-4bb8-8d0b-1c10f8a753ba", + "email": "host@example.com", + "nickname": "Host", + "name": "Host", + "password": "123456", + "createdAt": "2026-07-28T15:53:26Z" + } +] \ No newline at end of file diff --git a/linux-native-backend-v1-usable/docker-compose.yml b/linux-native-backend-v1-usable/docker-compose.yml new file mode 100644 index 0000000..9f94a94 --- /dev/null +++ b/linux-native-backend-v1-usable/docker-compose.yml @@ -0,0 +1,12 @@ +services: + brass-backend: + image: brass-birmingham-backend:latest + container_name: brass-birmingham-backend + restart: unless-stopped + ports: + - "8765:8765" + volumes: + - /mnt/disk2/dockdata/data:/app/data + environment: + HOST: 0.0.0.0 + PORT: 8765 diff --git a/linux-native-backend-v1-usable/run.sh b/linux-native-backend-v1-usable/run.sh new file mode 100644 index 0000000..eed78f4 --- /dev/null +++ b/linux-native-backend-v1-usable/run.sh @@ -0,0 +1,10 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +HOST="${HOST:-0.0.0.0}" +PORT="${PORT:-8765}" + +cd "$SCRIPT_DIR" +exec python3 -m app.server --host "$HOST" --port "$PORT" diff --git a/linux-native-backend-v1-usable/test_local.py b/linux-native-backend-v1-usable/test_local.py new file mode 100644 index 0000000..a1068da --- /dev/null +++ b/linux-native-backend-v1-usable/test_local.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import json +import threading +import time +import urllib.error +import urllib.request + +from app.server import build_server + + +def call(method: str, url: str, payload: dict | None = None, headers: dict | None = None) -> tuple[int, dict]: + data = None + req_headers = {"Content-Type": "application/json"} + if headers: + req_headers.update(headers) + if payload is not None: + data = json.dumps(payload).encode("utf-8") + + request = urllib.request.Request(url, data=data, method=method, headers=req_headers) + try: + with urllib.request.urlopen(request) as response: + return response.status, json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + +def main() -> None: + server = build_server("127.0.0.1", 18765) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + time.sleep(0.2) + + base = "http://127.0.0.1:18765" + status, payload = call( + "POST", + f"{base}/register", + {"email": "host@example.com", "password": "123456", "name": "Host"}, + ) + assert status in {201, 409}, (status, payload) + + status, payload = call( + "POST", + f"{base}/login", + {"email": "host@example.com", "password": "123456"}, + ) + assert status == 200, (status, payload) + token = payload["access_token"] + headers = {"Authorization": f"Bearer {token}"} + + status, payload = call("GET", f"{base}/me", headers=headers) + assert status == 200, (status, payload) + + status, payload = call( + "POST", + f"{base}/1.0.0/games", + { + "GameId": "test-game-001", + "GameStartConfiguration": {"NumberOfPlayersAsInt": 4, "IsPrivate": False}, + }, + headers=headers, + ) + assert status in {201, 409}, (status, payload) + + status, payload = call("GET", f"{base}/1.0.0/games?perPage=20", headers=headers) + assert status == 200 and isinstance(payload.get("data"), list), (status, payload) + + status, payload = call( + "POST", + f"{base}/1.0.0/games/test-game-001/events", + {"events": [{"$type": "GameStarted"}]}, + headers=headers, + ) + assert status == 200 and isinstance(payload, dict), (status, payload) + + status, payload = call( + "GET", + f"{base}/1.0.0/games/test-game-001/events-and-metadata", + headers=headers, + ) + assert status == 200, (status, payload) + + server.shutdown() + print("local test passed") + + +if __name__ == "__main__": + main() diff --git a/linux-native-backend-v1-usable/test_turn_progression.py b/linux-native-backend-v1-usable/test_turn_progression.py new file mode 100644 index 0000000..43a1bb8 --- /dev/null +++ b/linux-native-backend-v1-usable/test_turn_progression.py @@ -0,0 +1,72 @@ +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from app.server import sync_progress_from_events + + +class TurnProgressionTests(unittest.TestCase): + def test_turn_ended_advances_to_next_player(self) -> None: + metadata = { + "GameStartConfiguration": { + "NumberOfPlayersAsInt": 2, + "Players": { + "$values": [ + {"Color": "Yellow", "SittingOrder": 0}, + {"Color": "Blue", "SittingOrder": 1}, + ] + }, + }, + "PlayerMetadata": [ + {"PlayerNetworkId": "player-yellow", "NickName": "Yellow"}, + {"PlayerNetworkId": "player-blue", "NickName": "Blue"}, + ], + "NetworkGameProgressInformation": {}, + } + game = { + "metadata": metadata, + "events": [ + {"$type": "BoardGameRules.Entities.GamePhases.Events.TurnStarted, Assembly-CSharp", "PlayerColor": "Yellow"}, + {"$type": "BoardGameRules.Entities.GamePhases.Events.TurnEnded, Assembly-CSharp"}, + ], + } + + sync_progress_from_events(game) + + self.assertEqual(metadata["NetworkGameProgressInformation"]["CurrentPlayerId"], "player-blue") + + def test_turn_started_after_turn_ended_is_accepted(self) -> None: + metadata = { + "GameStartConfiguration": { + "NumberOfPlayersAsInt": 2, + "Players": { + "$values": [ + {"Color": "Yellow", "SittingOrder": 0}, + {"Color": "Blue", "SittingOrder": 1}, + ] + }, + }, + "PlayerMetadata": [ + {"PlayerNetworkId": "player-yellow", "NickName": "Yellow"}, + {"PlayerNetworkId": "player-blue", "NickName": "Blue"}, + ], + "NetworkGameProgressInformation": {}, + } + game = { + "metadata": metadata, + "events": [ + {"$type": "BoardGameRules.Entities.GamePhases.Events.TurnStarted, Assembly-CSharp", "PlayerColor": "Yellow"}, + {"$type": "BoardGameRules.Entities.GamePhases.Events.TurnEnded, Assembly-CSharp"}, + {"$type": "BoardGameRules.Entities.GamePhases.Events.TurnStarted, Assembly-CSharp", "PlayerColor": "Blue"}, + ], + } + + sync_progress_from_events(game) + + self.assertEqual(metadata["NetworkGameProgressInformation"]["CurrentPlayerId"], "player-blue") + + +if __name__ == "__main__": + unittest.main()