diff --git a/server/ban_store.py b/server/ban_store.py index 7747bbb..43fce70 100644 --- a/server/ban_store.py +++ b/server/ban_store.py @@ -1,12 +1,11 @@ -""" -Persistent IP ban list for Commonwealth Online servers. - -Stored as JSON next to the server config (default: bans.json). -""" +"""Persistent, fail-loud IP ban storage for Commonwealth Online servers.""" from __future__ import annotations +import ipaddress import json +import os +import tempfile import threading import time from dataclasses import dataclass @@ -14,35 +13,39 @@ from pathlib import Path from typing import Any -@dataclass +@dataclass(frozen=True) 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, - } + 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: + raw_ip = str(data.get("ip", "")).strip() + if not raw_ip: return None - reason = str(data.get("reason", "") or "") + try: + normalized_ip = str(ipaddress.ip_address(raw_ip)) + except ValueError: + return None + reason = str(data.get("reason", "") or "")[:1024] 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) + return cls(ip=normalized_ip, reason=reason, banned_at=banned_at_f) class BanStore: - """Thread-safe IP ban list with JSON persistence.""" + """Thread-safe IP ban list with atomic JSON persistence. + + Existing ban files are treated as security state. If one exists but cannot be parsed, + startup fails instead of silently replacing the effective ban list with an empty one. + """ def __init__(self, path: str | Path | None = None) -> None: self._path = Path(path) if path is not None else None @@ -59,28 +62,44 @@ class BanStore: with self._lock: self._path = Path(path) if path is not None else None + @staticmethod + def _normalize_ip(ip: str) -> str: + text = str(ip).strip() + if not text: + raise ValueError("IP address cannot be empty.") + try: + return str(ipaddress.ip_address(text)) + except ValueError as error: + raise ValueError(f"Invalid IP address: {text!r}") from error + 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: + with self._path.open("r", encoding="utf-8") as handle: data = json.load(handle) - except (OSError, json.JSONDecodeError): - self._bans = {} - return + except (OSError, UnicodeError, json.JSONDecodeError) as error: + raise RuntimeError(f"Could not load ban file {self._path}: {error}") from error + if not isinstance(data, dict) or not isinstance(data.get("banned_ips", []), list): + raise RuntimeError(f"Ban file {self._path} has an invalid schema.") 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 + invalid_entries = 0 + for item in data.get("banned_ips", []): + if not isinstance(item, dict): + invalid_entries += 1 + continue + entry = BanEntry.from_dict(item) + if entry is None: + invalid_entries += 1 + continue + bans[entry.ip] = entry + if invalid_entries: + raise RuntimeError( + f"Ban file {self._path} contains {invalid_entries} invalid entr{'y' if invalid_entries == 1 else 'ies'}." + ) self._bans = bans def save(self) -> None: @@ -89,39 +108,66 @@ class BanStore: 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)] + "banned_ips": [ + entry.to_dict() + for entry in sorted(self._bans.values(), key=lambda entry: entry.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") + fd, temporary_name = tempfile.mkstemp( + prefix=f".{self._path.name}.", + suffix=".tmp", + dir=str(self._path.parent), + text=True, + ) + temporary_path = Path(temporary_name) + try: + if os.name != "nt": + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle: + json.dump(payload, handle, indent=2, ensure_ascii=False, allow_nan=False) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, self._path) + if os.name != "nt": + os.chmod(self._path, 0o600) + except Exception: + try: + os.close(fd) + except OSError: + pass + try: + temporary_path.unlink(missing_ok=True) + except OSError: + pass + raise def is_banned(self, ip: str) -> bool: - normalized = str(ip).strip() + try: + normalized = self._normalize_ip(ip) + except ValueError: + return False with self._lock: return normalized in self._bans def get_ban(self, ip: str) -> BanEntry | None: - normalized = str(ip).strip() + try: + normalized = self._normalize_ip(ip) + except ValueError: + return None 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(), - ) + normalized = self._normalize_ip(ip) + entry = BanEntry(ip=normalized, reason=str(reason or "")[:1024], 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() + normalized = self._normalize_ip(ip) with self._lock: if normalized not in self._bans: return False @@ -131,4 +177,7 @@ class BanStore: 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)] + return [ + BanEntry(ip=entry.ip, reason=entry.reason, banned_at=entry.banned_at) + for entry in sorted(self._bans.values(), key=lambda entry: entry.ip) + ]