#!/usr/bin/env python3 import base64 import hashlib import hmac import html import json import os import secrets import sqlite3 import threading import time from http import HTTPStatus from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse BASE_DIR = Path(__file__).resolve().parent DB_PATH = Path(os.environ.get("COALGOV_DB", "/opt/mcworldgen-server/data/plugins/CoalGov/coalgov.db")) SECRET_PATH = Path(os.environ.get("COALGOV_WEB_SECRET", str(BASE_DIR / "session.secret"))) HOST = os.environ.get("COALGOV_WEB_HOST", "0.0.0.0") PORT = int(os.environ.get("COALGOV_WEB_PORT", "8088")) SESSION_TTL = int(os.environ.get("COALGOV_WEB_SESSION_SECONDS", "43200")) BLUEMAP_URL = os.environ.get("COALGOV_BLUEMAP_URL", "") BLUEMAP_MARKERS_PATH = os.environ.get("COALGOV_BLUEMAP_MARKERS_PATH", "") BLUEMAP_SYNC_SECONDS = int(os.environ.get("COALGOV_BLUEMAP_SYNC_SECONDS", "30")) def secret(): if SECRET_PATH.exists(): return SECRET_PATH.read_bytes().strip() value = secrets.token_bytes(32) SECRET_PATH.write_bytes(base64.urlsafe_b64encode(value)) os.chmod(SECRET_PATH, 0o600) return SECRET_PATH.read_bytes().strip() SECRET = secret() def db(): con = sqlite3.connect(DB_PATH) con.row_factory = sqlite3.Row return con def now_ms(): return int(time.time() * 1000) def token_hash(token): return hashlib.sha256(token.upper().replace(" ", "").encode()).hexdigest() def sign(payload): raw = json.dumps(payload, separators=(",", ":")).encode() body = base64.urlsafe_b64encode(raw).decode().rstrip("=") sig = hmac.new(SECRET, body.encode(), hashlib.sha256).hexdigest() return f"{body}.{sig}" def unsign(value): if not value or "." not in value: return None body, sig = value.rsplit(".", 1) expected = hmac.new(SECRET, body.encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(sig, expected): return None try: raw = base64.urlsafe_b64decode(body + "=" * (-len(body) % 4)) payload = json.loads(raw) except (ValueError, json.JSONDecodeError): return None if payload.get("exp", 0) < int(time.time()): return None return payload def money(cents): cents = int(cents or 0) coal, cc = divmod(abs(cents), 100) prefix = "-" if cents < 0 else "" if coal and cc: return f"{prefix}{coal} coal {cc}cc" if coal: return f"{prefix}{coal} coal" return f"{prefix}{cc}cc" def rows(query, args=()): with db() as con: return [dict(row) for row in con.execute(query, args).fetchall()] def row(query, args=()): with db() as con: found = con.execute(query, args).fetchone() return dict(found) if found else None def display_claim(claim): name = claim.get("display_name") or f"Claim #{claim['id']}" return {**claim, "display": name, "tax_due_text": money(claim.get("tax_due", 0))} def can_manage(con, uuid, claim_id): claim = con.execute("SELECT owner_uuid FROM claims WHERE id = ?", (claim_id,)).fetchone() if not claim: return False if claim["owner_uuid"] == uuid: return True grant = con.execute(""" SELECT 1 FROM claim_permissions WHERE claim_id = ? AND player_uuid = ? AND permission = 'MANAGE' LIMIT 1 """, (claim_id, uuid)).fetchone() return bool(grant) def color(kind): colors = { "HOMESTEAD": {"r": 102, "g": 170, "b": 204}, "INDUSTRIAL": {"r": 204, "g": 153, "b": 102}, "GOVERNMENT": {"r": 221, "g": 102, "b": 102}, "PUBLIC": {"r": 119, "g": 204, "b": 119}, "PROTECTED_PRESERVE": {"r": 102, "g": 204, "b": 102}, "MINING_CONCESSION": {"r": 204, "g": 204, "b": 102}, "BORDER_ZONE": {"r": 204, "g": 102, "b": 204}, } return colors.get(kind, {"r": 170, "g": 170, "b": 170}) def rect_shape(area): x1, x2 = sorted((int(area["x1"]), int(area["x2"]))) z1, z2 = sorted((int(area["z1"]), int(area["z2"]))) return [ {"x": x1, "z": z1}, {"x": x2 + 1, "z": z1}, {"x": x2 + 1, "z": z2 + 1}, {"x": x1, "z": z2 + 1}, ] def center(shape): return { "x": sum(p["x"] for p in shape) / len(shape), "y": 72, "z": sum(p["z"] for p in shape) / len(shape), } def marker_detail(title, lines): body = "".join(f"
{html.escape(str(line))}
" for line in lines) return f"{html.escape(str(title))}{body}" def build_bluemap_markers(): with db() as con: claims = [display_claim(dict(r)) for r in con.execute(""" SELECT c.*, p.name AS owner_name FROM claims c LEFT JOIN players p ON p.uuid = c.owner_uuid WHERE c.world = 'world' ORDER BY c.id """)] vertices = {} for r in con.execute("SELECT claim_id, x, z FROM claim_vertices ORDER BY claim_id, vertex_order"): vertices.setdefault(r["claim_id"], []).append({"x": int(r["x"]), "z": int(r["z"])}) lands = [dict(r) for r in con.execute("SELECT * FROM land_regions WHERE world = 'world' ORDER BY name")] claim_markers = [] for claim in claims: shape = vertices.get(claim["id"]) or rect_shape(claim) c = color(claim["claim_type"]) claim_markers.append({ "id": f"claim-{claim['id']}", "type": "shape", "map": "overworld", "position": center(shape), "label": claim["display"], "detail": marker_detail(claim["display"], [ f"Type: {claim['claim_type']}", f"Owner: {claim.get('owner_name') or claim['owner_uuid']}", f"Tax due: {claim['tax_due_text']}", ]), "shape": shape, "shapeY": 72, "depthTest": False, "lineWidth": 3, "lineColor": {**c, "a": 1.0}, "fillColor": {**c, "a": 0.28}, "minDistance": 10.0, "maxDistance": 10000000.0, }) land_markers = [] for land in lands: shape = rect_shape(land) c = color(land["land_class"]) land_markers.append({ "id": f"land-{land['id']}", "type": "shape", "map": "overworld", "position": center(shape), "label": land["name"], "detail": marker_detail(land["name"], [f"Class: {land['land_class']}"]), "shape": shape, "shapeY": 73, "depthTest": False, "lineWidth": 2, "lineColor": {**c, "a": 1.0}, "fillColor": {**c, "a": 0.16}, "minDistance": 10.0, "maxDistance": 10000000.0, }) return { "coalgov-claims": { "label": "CoalGov Claims", "toggleable": True, "defaultHidden": False, "sorting": 10, "markers": {marker["id"]: {k: v for k, v in marker.items() if k != "id"} for marker in claim_markers}, }, "coalgov-land": { "label": "CoalGov Land", "toggleable": True, "defaultHidden": False, "sorting": 11, "markers": {marker["id"]: {k: v for k, v in marker.items() if k != "id"} for marker in land_markers}, }, } def sync_bluemap_markers_once(): if not BLUEMAP_MARKERS_PATH: return False target = Path(BLUEMAP_MARKERS_PATH) target.parent.mkdir(parents=True, exist_ok=True) tmp = target.with_suffix(target.suffix + ".tmp") tmp.write_text(json.dumps(build_bluemap_markers(), separators=(",", ":"))) tmp.replace(target) return True def sync_bluemap_markers_loop(): while True: try: sync_bluemap_markers_once() except Exception as exc: print(f"BlueMap marker sync failed: {exc}", flush=True) time.sleep(max(5, BLUEMAP_SYNC_SECONDS)) INDEX = """ CoalGov Property Map

CoalGov Property Map

""".replace("__BLUEMAP_URL__", html.escape(BLUEMAP_URL, quote=True)) class Handler(BaseHTTPRequestHandler): def do_GET(self): path = urlparse(self.path).path if path == "/": return self.html(INDEX) if path == "/api/me": user = self.require_user() if user: player = row("SELECT name FROM players WHERE uuid = ?", (user["uuid"],)) or {} self.json({"uuid": user["uuid"], "name": player.get("name", user["uuid"])}) return if path == "/api/map": user = self.require_user() if not user: return uuid = user["uuid"] with db() as con: claims = [display_claim(dict(r)) for r in con.execute(""" SELECT c.*, p.name AS owner_name FROM claims c LEFT JOIN players p ON p.uuid = c.owner_uuid ORDER BY c.id """)] vertices = {} for r in con.execute("SELECT claim_id, x, z FROM claim_vertices ORDER BY claim_id, vertex_order"): vertices.setdefault(r["claim_id"], []).append({"x": r["x"], "z": r["z"]}) for c in claims: c["vertices"] = vertices.get(c["id"], []) c["manageable"] = c["owner_uuid"] == uuid or bool(con.execute(""" SELECT 1 FROM claim_permissions WHERE claim_id = ? AND player_uuid = ? AND permission = 'MANAGE' """, (c["id"], uuid)).fetchone()) lands = [dict(r) for r in con.execute("SELECT * FROM land_regions ORDER BY name")] self.json({"claims": claims, "lands": lands}) return self.error(HTTPStatus.NOT_FOUND, "Not found") def do_POST(self): path = urlparse(self.path).path if path == "/api/login": body = self.body() token = body.get("token", "") hashed = token_hash(token) with db() as con: found = con.execute(""" SELECT token_hash, player_uuid FROM web_tokens WHERE token_hash = ? AND expires_at > ? AND used_at IS NULL """, (hashed, now_ms())).fetchone() if not found: return self.error(HTTPStatus.UNAUTHORIZED, "Invalid or expired token") con.execute("UPDATE web_tokens SET used_at = ? WHERE token_hash = ?", (now_ms(), hashed)) con.commit() cookie = f"cg_session={sign({'uuid': found['player_uuid'], 'exp': int(time.time()) + SESSION_TTL})}; HttpOnly; SameSite=Lax; Path=/" self.json({"ok": True}, headers={"Set-Cookie": cookie}) return if path == "/api/logout": self.json({"ok": True}, headers={"Set-Cookie": "cg_session=; Max-Age=0; Path=/"}) return user = self.require_user() if not user: return if path == "/api/bluemap/sync": if sync_bluemap_markers_once(): return self.json({"ok": True}) return self.error(HTTPStatus.BAD_REQUEST, "BlueMap marker sync is not configured") if path.startswith("/api/claims/") and path.endswith("/rename"): claim_id = int(path.split("/")[3]) name = str(self.body().get("name", "")).strip()[:48] if not name: return self.error(HTTPStatus.BAD_REQUEST, "Name required") with db() as con: if not can_manage(con, user["uuid"], claim_id): return self.error(HTTPStatus.FORBIDDEN, "You cannot manage that claim") con.execute("UPDATE claims SET display_name = ? WHERE id = ?", (name, claim_id)) con.commit() return self.json({"ok": True}) if path.startswith("/api/claims/") and path.endswith("/transfer"): claim_id = int(path.split("/")[3]) target = str(self.body().get("target", "")).strip() with db() as con: if not can_manage(con, user["uuid"], claim_id): return self.error(HTTPStatus.FORBIDDEN, "You cannot manage that claim") player = con.execute("SELECT uuid, name FROM players WHERE uuid = ? OR lower(name) = lower(?)", (target, target)).fetchone() if not player: return self.error(HTTPStatus.BAD_REQUEST, "Target player not found in CoalGov") con.execute("UPDATE claims SET owner_uuid = ? WHERE id = ?", (player["uuid"], claim_id)) con.execute("DELETE FROM claim_permissions WHERE claim_id = ?", (claim_id,)) con.commit() return self.json({"ok": True}) if path.startswith("/api/claims/") and path.endswith("/pay-tax"): claim_id = int(path.split("/")[3]) with db() as con: if not can_manage(con, user["uuid"], claim_id): return self.error(HTTPStatus.FORBIDDEN, "You cannot manage that claim") claim = con.execute("SELECT tax_due FROM claims WHERE id = ?", (claim_id,)).fetchone() due = int(claim["tax_due"] or 0) if claim else 0 if due <= 0: return self.json({"ok": True, "message": "No tax due."}) updated = con.execute("UPDATE players SET coal_balance = coal_balance - ? WHERE uuid = ? AND coal_balance >= ?", (due, user["uuid"], due)).rowcount if updated != 1: return self.error(HTTPStatus.BAD_REQUEST, f"Balance too low. Need {money(due)}.") con.execute("UPDATE claims SET tax_due = 0 WHERE id = ?", (claim_id,)) con.execute("INSERT INTO transactions(from_uuid,to_uuid,amount,reason,created_at) VALUES (?,?,?,?,?)", (user["uuid"], None, due, "claim_tax_web", now_ms())) con.execute("INSERT INTO treasury(id,balance) VALUES ('main', ?) ON CONFLICT(id) DO UPDATE SET balance = balance + excluded.balance", (due,)) con.commit() return self.json({"ok": True, "message": f"Paid {money(due)}."}) self.error(HTTPStatus.NOT_FOUND, "Not found") def require_user(self): cookie = SimpleCookie(self.headers.get("Cookie", "")) payload = unsign(cookie.get("cg_session").value if cookie.get("cg_session") else "") if not payload: self.error(HTTPStatus.UNAUTHORIZED, "Login required") return None return payload def body(self): length = int(self.headers.get("Content-Length", "0")) if length <= 0: return {} return json.loads(self.rfile.read(length) or b"{}") def html(self, content): raw = content.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def json(self, value, headers=None): raw = json.dumps(value).encode() self.send_response(200) for key, val in (headers or {}).items(): self.send_header(key, val) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def error(self, status, message): raw = json.dumps({"error": message}).encode() self.send_response(status.value) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) if __name__ == "__main__": print(f"CoalGov property web listening on {HOST}:{PORT}, db={DB_PATH}") if BLUEMAP_MARKERS_PATH: threading.Thread(target=sync_bluemap_markers_loop, daemon=True).start() ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()