Complete TASK-072 RCON provider setup

backend/app/config.py

@@ -12,6 +12,7 @@ DEFAULT_STORAGE_FILENAME = "hll_vietnam_dev.sqlite3"
 DEFAULT_REFRESH_INTERVAL_SECONDS = 300
 DEFAULT_LIVE_DATA_SOURCE = "a2s"
 DEFAULT_HISTORICAL_DATA_SOURCE = "public-scoreboard"
+DEFAULT_RCON_TIMEOUT_SECONDS = 10.0
 DEFAULT_HISTORICAL_CRCON_PAGE_SIZE = 50
 DEFAULT_HISTORICAL_CRCON_TIMEOUT_SECONDS = 15.0
 DEFAULT_HISTORICAL_CRCON_DETAIL_WORKERS = 8
@@ -33,6 +34,8 @@ DEFAULT_ALLOWED_ORIGINS = (
 )
 DEFAULT_A2S_TARGETS_ENV_VAR = "HLL_BACKEND_A2S_TARGETS"
 DEFAULT_A2S_SOURCE_NAME = "community-hispana-a2s"
+DEFAULT_RCON_TARGETS_ENV_VAR = "HLL_BACKEND_RCON_TARGETS"
+DEFAULT_RCON_SOURCE_NAME = "community-hispana-rcon"
 
 
 def get_bind_address() -> tuple[str, int]:
@@ -187,6 +190,18 @@ def get_historical_data_source_kind() -> str:
     return source_kind
 
 
+def get_rcon_request_timeout_seconds() -> float:
+    """Return the timeout used for HLL RCON TCP requests."""
+    configured_value = os.getenv(
+        "HLL_BACKEND_RCON_TIMEOUT_SECONDS",
+        str(DEFAULT_RCON_TIMEOUT_SECONDS),
+    )
+    timeout_seconds = float(configured_value)
+    if timeout_seconds <= 0:
+        raise ValueError("HLL_BACKEND_RCON_TIMEOUT_SECONDS must be positive.")
+    return timeout_seconds
+
+
 def get_historical_refresh_max_retries() -> int:
     """Return the retry count used by the historical refresh loop."""
     configured_value = os.getenv(
@@ -262,3 +277,13 @@ def get_a2s_targets_payload() -> str | None:
 
     normalized = raw_payload.strip()
     return normalized or None
+
+
+def get_rcon_targets_payload() -> str | None:
+    """Return the optional JSON payload that defines live RCON targets."""
+    raw_payload = os.getenv(DEFAULT_RCON_TARGETS_ENV_VAR)
+    if raw_payload is None:
+        return None
+
+    normalized = raw_payload.strip()
+    return normalized or None

backend/app/data_sources.py

@@ -8,6 +8,7 @@ from typing import Protocol
 from .collector import collect_server_snapshots
 from .config import get_historical_data_source_kind, get_live_data_source_kind
 from .providers.public_scoreboard_provider import PublicScoreboardHistoricalDataSource
+from .providers.rcon_provider import RconLiveDataSource
 from .server_targets import A2SServerTarget, load_a2s_targets
 
 
@@ -91,19 +92,6 @@ class RconHistoricalDataSource:
         raise RuntimeError("Historical RCON provider is not implemented yet.")
 
 
-@dataclass(frozen=True, slots=True)
-class RconLiveDataSource:
-    """Placeholder live provider for future production RCON integration."""
-
-    source_kind: str = SOURCE_KIND_RCON
-
-    def collect_snapshots(self, *, persist: bool) -> dict[str, object]:
-        raise RuntimeError("Live RCON provider is not implemented yet.")
-
-    def build_target_index(self) -> dict[str | None, A2SServerTarget]:
-        return {}
-
-
 def get_historical_data_source() -> HistoricalDataSource:
     """Select the historical provider configured for the current environment."""
     source_kind = get_historical_data_source_kind()

backend/app/payloads.py

@@ -4,7 +4,11 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 
-from .config import get_refresh_interval_seconds
+from .config import (
+    get_historical_data_source_kind,
+    get_live_data_source_kind,
+    get_refresh_interval_seconds,
+)
 from .data_sources import get_live_data_source
 from .historical_snapshot_storage import get_historical_snapshot
 from .historical_snapshots import (
@@ -36,6 +40,8 @@ def build_health_payload() -> dict[str, str]:
         "status": "ok",
         "service": "hll-vietnam-backend",
         "phase": "bootstrap",
+        "live_data_source": get_live_data_source_kind(),
+        "historical_data_source": get_historical_data_source_kind(),
     }
 
 

backend/app/providers/rcon_provider.py

@@ -0,0 +1,63 @@
+"""RCON provider adapter for live HLL server state."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ..rcon_client import RconServerTarget, load_rcon_targets, query_live_server_state
+from ..snapshots import build_snapshot_batch, utc_now
+from ..storage import persist_snapshot_batch
+
+
+@dataclass(frozen=True, slots=True)
+class RconLiveDataSource:
+    """Live provider backed by direct HLL RCON access."""
+
+    source_kind: str = "rcon"
+
+    def collect_snapshots(self, *, persist: bool) -> dict[str, object]:
+        configured_targets = load_rcon_targets()
+        if not configured_targets:
+            raise RuntimeError("No RCON targets configured in HLL_BACKEND_RCON_TARGETS.")
+
+        captured_at = utc_now()
+        normalized_records: list[dict[str, object]] = []
+        errors: list[dict[str, object]] = []
+
+        for target in configured_targets:
+            try:
+                normalized_records.append(query_live_server_state(target))
+            except Exception as error:  # noqa: BLE001 - keep provider failures controlled
+                errors.append(
+                    {
+                        "target": target.name,
+                        "host": target.host,
+                        "port": target.port,
+                        "message": str(error),
+                    }
+                )
+
+        payload = {
+            "source_name": "hll-rcon",
+            "collection_mode": "rcon",
+            "fallback_used": False,
+            "target_count": len(configured_targets),
+            "success_count": len(normalized_records),
+            "errors": errors,
+            "captured_at": captured_at.isoformat().replace("+00:00", "Z"),
+            "snapshots": build_snapshot_batch(normalized_records, captured_at=captured_at),
+        }
+        if persist:
+            payload["storage"] = persist_snapshot_batch(
+                payload["snapshots"],
+                source_name=payload["source_name"],
+                captured_at=payload["captured_at"],
+            )
+        return payload
+
+    def build_target_index(self) -> dict[str | None, RconServerTarget]:
+        return {
+            target.external_server_id: target
+            for target in load_rcon_targets()
+            if target.external_server_id
+        }

backend/app/rcon_client.py

@@ -0,0 +1,184 @@
+"""Minimal Hell Let Loose RCON client for live server state queries."""
+
+from __future__ import annotations
+
+import json
+import socket
+from dataclasses import dataclass
+
+from .config import (
+    DEFAULT_RCON_SOURCE_NAME,
+    get_rcon_request_timeout_seconds,
+    get_rcon_targets_payload,
+)
+
+
+RCON_BUFFER_SIZE = 32768
+
+
+@dataclass(frozen=True, slots=True)
+class RconServerTarget:
+    """Configuration needed to query one HLL RCON endpoint."""
+
+    name: str
+    host: str
+    port: int
+    password: str
+    source_name: str
+    external_server_id: str | None = None
+    region: str | None = None
+    game_port: int | None = None
+    query_port: int | None = None
+
+
+class HllRconConnection:
+    """Tiny synchronous HLL RCON connection using the documented XOR flow."""
+
+    def __init__(self, *, timeout_seconds: float) -> None:
+        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._socket.settimeout(timeout_seconds)
+        self._xor_key: bytes | None = None
+
+    def connect(self, *, host: str, port: int, password: str) -> None:
+        self._socket.connect((host, port))
+        self._xor_key = self._socket.recv(RCON_BUFFER_SIZE)
+        response = self.execute(f"Login {password}")
+        if response != "SUCCESS":
+            raise RuntimeError("Invalid RCON password.")
+
+    def execute(self, command: str) -> str:
+        payload = command.encode("utf-8")
+        self._socket.sendall(self._xor(payload))
+        return self._receive().decode("utf-8", errors="replace")
+
+    def close(self) -> None:
+        try:
+            self._socket.shutdown(socket.SHUT_RDWR)
+        except OSError:
+            pass
+        self._socket.close()
+
+    def _receive(self) -> bytes:
+        chunks: list[bytes] = []
+        while True:
+            chunk = self._socket.recv(RCON_BUFFER_SIZE)
+            chunks.append(self._xor(chunk))
+            if len(chunk) < RCON_BUFFER_SIZE:
+                break
+        return b"".join(chunks)
+
+    def _xor(self, payload: bytes) -> bytes:
+        if not self._xor_key:
+            raise RuntimeError("The HLL server did not provide an RCON XOR key.")
+        return bytes(
+            value ^ self._xor_key[index % len(self._xor_key)]
+            for index, value in enumerate(payload)
+        )
+
+    def __enter__(self) -> HllRconConnection:
+        return self
+
+    def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
+        self.close()
+
+
+def load_rcon_targets() -> tuple[RconServerTarget, ...]:
+    """Load RCON targets from JSON env payload."""
+    raw_payload = get_rcon_targets_payload()
+    if raw_payload is None:
+        return ()
+    parsed = json.loads(raw_payload)
+    if not isinstance(parsed, list):
+        raise ValueError("HLL_BACKEND_RCON_TARGETS must be a JSON array.")
+    return tuple(_coerce_rcon_target(item) for item in parsed if isinstance(item, dict))
+
+
+def query_live_server_state(
+    target: RconServerTarget,
+    *,
+    timeout_seconds: float | None = None,
+) -> dict[str, object]:
+    """Query one HLL server via RCON and normalize it to the live snapshot shape."""
+    resolved_timeout = timeout_seconds or get_rcon_request_timeout_seconds()
+    with HllRconConnection(timeout_seconds=resolved_timeout) as connection:
+        connection.connect(host=target.host, port=target.port, password=target.password)
+        server_name = connection.execute("Get Name").strip()
+        slots = connection.execute("Get Slots").strip()
+        game_state = connection.execute("Get GameState").strip()
+
+    players, max_players = _parse_slots(slots)
+    current_map = _parse_gamestate_value(game_state, "Map")
+    resolved_external_id = target.external_server_id or f"rcon:{target.host}:{target.port}"
+    return {
+        "external_server_id": resolved_external_id,
+        "server_name": server_name or target.name,
+        "status": "online",
+        "players": players,
+        "max_players": max_players,
+        "current_map": current_map,
+        "region": target.region,
+        "source_name": target.source_name,
+        "snapshot_origin": "real-rcon",
+        "source_ref": f"rcon://{target.host}:{target.port}",
+    }
+
+
+def _coerce_rcon_target(raw_target: dict[str, object]) -> RconServerTarget:
+    name = str(raw_target.get("name") or "Unnamed RCON target").strip()
+    host = str(raw_target.get("host") or "").strip()
+    password = str(raw_target.get("password") or "").strip()
+    source_name = str(raw_target.get("source_name") or DEFAULT_RCON_SOURCE_NAME).strip()
+    port = int(raw_target.get("port") or 0)
+    if not host:
+        raise ValueError("Each RCON target must define a non-empty host.")
+    if port <= 0:
+        raise ValueError("Each RCON target must define a valid port.")
+    if not password:
+        raise ValueError("Each RCON target must define a non-empty password.")
+
+    return RconServerTarget(
+        name=name,
+        host=host,
+        port=port,
+        password=password,
+        source_name=source_name or DEFAULT_RCON_SOURCE_NAME,
+        external_server_id=_string_or_none(raw_target.get("external_server_id")),
+        region=_string_or_none(raw_target.get("region")),
+        game_port=_coerce_optional_positive_int(raw_target.get("game_port")),
+        query_port=_coerce_optional_positive_int(raw_target.get("query_port")),
+    )
+
+
+def _parse_slots(payload: str) -> tuple[int | None, int | None]:
+    if "/" not in payload:
+        return None, None
+    left, right = payload.split("/", 1)
+    try:
+        return int(left.strip()), int(right.strip())
+    except ValueError:
+        return None, None
+
+
+def _parse_gamestate_value(payload: str, label: str) -> str | None:
+    prefix = f"{label}:"
+    for line in payload.splitlines():
+        if line.startswith(prefix):
+            value = line.removeprefix(prefix).strip()
+            return value or None
+    return None
+
+
+def _string_or_none(value: object) -> str | None:
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip()
+    return normalized or None
+
+
+def _coerce_optional_positive_int(value: object) -> int | None:
+    if value is None:
+        return None
+    coerced = int(value)
+    if coerced <= 0:
+        raise ValueError("Configured RCON target ports must be positive when defined.")
+    return coerced