andrew
881aa33eef
Adds end-to-end Linux compatibility work for the dedicated server: new CI workflow (Ubuntu + Arch), line-ending/executable safeguards, and a local venv-first startup flow with split dependency files for server vs optional host GUI tooling. Improves runtime resilience with better bind error messages, stricter config validation, writable-state checks, cleaner socket/thread shutdown behavior, and SIGTERM-aware graceful stop handling for headless/systemd use. Updates deployment/startup docs and adds portability/runtime integration tests to lock in these behaviors.
135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
"""
|
|
Persistent IP ban list for Commonwealth Online servers.
|
|
|
|
Stored as JSON next to the server config (default: bans.json).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class BanEntry:
|
|
ip: str
|
|
reason: str = ""
|
|
banned_at: float = 0.0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"ip": self.ip,
|
|
"reason": self.reason,
|
|
"bannedAt": self.banned_at,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> BanEntry | None:
|
|
ip = str(data.get("ip", "")).strip()
|
|
if not ip:
|
|
return None
|
|
reason = str(data.get("reason", "") or "")
|
|
banned_at = data.get("bannedAt", data.get("banned_at", 0.0))
|
|
try:
|
|
banned_at_f = float(banned_at) if banned_at is not None else 0.0
|
|
except (TypeError, ValueError):
|
|
banned_at_f = 0.0
|
|
return cls(ip=ip, reason=reason, banned_at=banned_at_f)
|
|
|
|
|
|
class BanStore:
|
|
"""Thread-safe IP ban list with JSON persistence."""
|
|
|
|
def __init__(self, path: str | Path | None = None) -> None:
|
|
self._path = Path(path) if path is not None else None
|
|
self._lock = threading.RLock()
|
|
self._bans: dict[str, BanEntry] = {}
|
|
if self._path is not None:
|
|
self.load()
|
|
|
|
@property
|
|
def path(self) -> Path | None:
|
|
return self._path
|
|
|
|
def set_path(self, path: str | Path | None) -> None:
|
|
with self._lock:
|
|
self._path = Path(path) if path is not None else None
|
|
|
|
def load(self) -> None:
|
|
with self._lock:
|
|
if self._path is None or not self._path.exists():
|
|
self._bans = {}
|
|
return
|
|
|
|
try:
|
|
with open(self._path, "r", encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except (OSError, json.JSONDecodeError):
|
|
self._bans = {}
|
|
return
|
|
|
|
bans: dict[str, BanEntry] = {}
|
|
raw_list = data.get("banned_ips", []) if isinstance(data, dict) else []
|
|
if isinstance(raw_list, list):
|
|
for item in raw_list:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
entry = BanEntry.from_dict(item)
|
|
if entry is not None:
|
|
bans[entry.ip] = entry
|
|
self._bans = bans
|
|
|
|
def save(self) -> None:
|
|
with self._lock:
|
|
if self._path is None:
|
|
return
|
|
self._path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = {
|
|
"banned_ips": [entry.to_dict() for entry in sorted(self._bans.values(), key=lambda e: e.ip)]
|
|
}
|
|
with open(self._path, "w", encoding="utf-8", newline="\n") as handle:
|
|
json.dump(payload, handle, indent=2, ensure_ascii=False)
|
|
handle.write("\n")
|
|
|
|
def is_banned(self, ip: str) -> bool:
|
|
normalized = str(ip).strip()
|
|
with self._lock:
|
|
return normalized in self._bans
|
|
|
|
def get_ban(self, ip: str) -> BanEntry | None:
|
|
normalized = str(ip).strip()
|
|
with self._lock:
|
|
return self._bans.get(normalized)
|
|
|
|
def ban_ip(self, ip: str, reason: str = "") -> BanEntry:
|
|
normalized = str(ip).strip()
|
|
if not normalized:
|
|
raise ValueError("IP address cannot be empty.")
|
|
|
|
entry = BanEntry(
|
|
ip=normalized,
|
|
reason=str(reason or ""),
|
|
banned_at=time.time(),
|
|
)
|
|
with self._lock:
|
|
self._bans[normalized] = entry
|
|
self.save()
|
|
return entry
|
|
|
|
def unban_ip(self, ip: str) -> bool:
|
|
normalized = str(ip).strip()
|
|
with self._lock:
|
|
if normalized not in self._bans:
|
|
return False
|
|
del self._bans[normalized]
|
|
self.save()
|
|
return True
|
|
|
|
def list_bans(self) -> list[BanEntry]:
|
|
with self._lock:
|
|
return [BanEntry(ip=e.ip, reason=e.reason, banned_at=e.banned_at) for e in sorted(self._bans.values(), key=lambda e: e.ip)]
|