server.py

"""RuneCards home-network server (standard-library only)."""

from __future__ import annotations

import argparse
import json
import mimetypes
import os
import socket
import threading
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import unquote, urlparse


ROOT = Path(__file__).resolve().parent
SAVE_DIR = Path(
    os.environ.get("RUNECARDS_DATA_DIR", Path.home() / ".runecards")
).expanduser().resolve()
SAVE_PATH = SAVE_DIR / "save.json"
SAVE_LOCK = threading.Lock()


class RuneCardsHandler(SimpleHTTPRequestHandler):
    server_version = "RuneCards/1.0"

    def __init__(self, *args, **kwargs):
        super().__init__(*args, directory=str(ROOT), **kwargs)

    def _json(self, status: int, payload: object) -> None:
        body = 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(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, message: str, *args: object) -> None:
        print(f"{self.address_string()} - {message % args}")

    @staticmethod
    def _is_public_path(raw_path: str) -> bool:
        path = unquote(urlparse(raw_path).path)
        allowed_files = {
            "/", "/index.html", "/styles.css", "/web-app.js",
            "/manifest.webmanifest", "/service-worker.js", "/config.json",
            "/cards.json", "/quests.json", "/assets/icons/icon-192.png",
            "/assets/icons/icon-512.png",
        }
        parts = Path(path).parts
        safe_asset = path.startswith("/assets/") and ".." not in parts
        return path in allowed_files or safe_asset

    def do_GET(self) -> None:
        path = urlparse(self.path).path
        if path == "/api/save":
            with SAVE_LOCK:
                if not SAVE_PATH.exists():
                    self._json(404, {"error": "No shared save exists yet"})
                    return
                try:
                    data = json.loads(SAVE_PATH.read_text(encoding="utf-8"))
                except (OSError, json.JSONDecodeError):
                    self._json(500, {"error": "The shared save could not be read"})
                    return
            self._json(200, data)
            return
        if not self._is_public_path(self.path):
            self.send_error(404, "Not found")
            return
        super().do_GET()

    def do_HEAD(self) -> None:
        if not self._is_public_path(self.path):
            self.send_error(404, "Not found")
            return
        super().do_HEAD()

    def do_PUT(self) -> None:
        if urlparse(self.path).path != "/api/save":
            self._json(404, {"error": "Not found"})
            return
        try:
            length = int(self.headers.get("Content-Length", "0"))
            if length <= 0 or length > 2_000_000:
                raise ValueError
            data = json.loads(self.rfile.read(length).decode("utf-8"))
            if not isinstance(data, dict):
                raise ValueError
        except (ValueError, UnicodeDecodeError, json.JSONDecodeError):
            self._json(400, {"error": "Invalid save data"})
            return

        with SAVE_LOCK:
            SAVE_DIR.mkdir(parents=True, exist_ok=True)
            temporary = SAVE_PATH.with_suffix(".json.tmp")
            temporary.write_text(
                json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
            )
            os.replace(temporary, SAVE_PATH)
        self._json(200, {"ok": True})

    def end_headers(self) -> None:
        self.send_header("X-RuneCards-Server", "1")
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("X-Frame-Options", "DENY")
        self.send_header("Referrer-Policy", "no-referrer")
        if urlparse(self.path).path in {"/", "/index.html", "/web-app.js", "/service-worker.js"}:
            self.send_header("Cache-Control", "no-cache")
        super().end_headers()


def local_addresses(port: int) -> list[str]:
    addresses = {"127.0.0.1"}
    try:
        addresses.update(socket.gethostbyname_ex(socket.gethostname())[2])
    except socket.gaierror:
        pass
    return [f"http://{address}:{port}" for address in sorted(addresses) if ":" not in address]


def main() -> None:
    parser = argparse.ArgumentParser(description="Serve RuneCards on your home network")
    parser.add_argument(
        "--host",
        default=os.environ.get("RUNECARDS_HOST", "0.0.0.0"),
        help="Interface to bind (default: all interfaces)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=int(os.environ.get("RUNECARDS_PORT", "8765")),
        help="TCP port (default: 8765)",
    )
    args = parser.parse_args()
    mimetypes.add_type("application/manifest+json", ".webmanifest")
    server = ThreadingHTTPServer((args.host, args.port), RuneCardsHandler)
    print("RuneCards is running. Keep this window open.")
    print(f"Shared save: {SAVE_PATH}")
    for address in local_addresses(args.port):
        print(f"Open: {address}")
    print("Press Ctrl+C to stop.")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nRuneCards stopped.")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()