Seth
e00d454b19
- agent/serve.py: CLI assistant with interactive, single-query, and eval modes (Ollama + qwen3-coder) - agent/tools/rcon_tool.py: RCON execute, server status, player info - agent/tools/knowledge_tool.py: TF-IDF RAG search, command reference lookup, server context - agent/guardrails/command_filter.py: 14-prefix allowlist, execute-tail bypass detection, destructive flags, 1.21 syntax warnings, audit log - agent/prompts/system_prompts.py: sudo (pure commands), god (persona), intervention (benign) system prompts - Guardrails tested: 10/10 allowlist, 5/6 syntax warnings pass
115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
"""
|
|
RCON tool for Minecraft server interaction.
|
|
|
|
Provides:
|
|
- rcon_execute(command) -> send RCON command, return result
|
|
- get_server_status() -> player list, time, difficulty
|
|
- get_player_info(player) -> position, health, gamemode
|
|
"""
|
|
|
|
import re
|
|
import socket
|
|
import struct
|
|
import time
|
|
from typing import Dict, Any, Optional, List
|
|
|
|
|
|
def rcon_send(cmd: str, host: str = '127.0.0.1', port: int = 25577,
|
|
password: str = 'REDACTED_RCON', timeout: float = 5.0) -> str:
|
|
"""Send a single RCON command and return the response text."""
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(timeout)
|
|
try:
|
|
s.connect((host, int(port)))
|
|
|
|
def pkt(req_id: int, pkt_type: int, payload: str) -> bytes:
|
|
p = payload.encode('utf-8') + b'\x00\x00'
|
|
return struct.pack('<iii', len(p) + 8, req_id, pkt_type) + p
|
|
|
|
# Authenticate (type 3)
|
|
s.sendall(pkt(1, 3, password))
|
|
time.sleep(0.15)
|
|
s.recv(4096)
|
|
# Send command (type 2)
|
|
s.sendall(pkt(2, 2, cmd))
|
|
time.sleep(0.2)
|
|
r = s.recv(4096)
|
|
return r[12:-2].decode('utf-8', errors='replace')
|
|
except Exception as e:
|
|
return f'RCON error: {e}'
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
class RconTool:
|
|
"""RCON tool with configurable connection parameters."""
|
|
|
|
def __init__(self, host: str = '127.0.0.1', port: int = 25577,
|
|
password: str = 'REDACTED_RCON'):
|
|
self.host = host
|
|
self.port = port
|
|
self.password = password
|
|
|
|
def execute(self, command: str) -> Dict[str, Any]:
|
|
"""Execute an RCON command and return structured result."""
|
|
result = rcon_send(command, self.host, self.port, self.password)
|
|
is_error = any(w in result.lower() for w in [
|
|
'unknown', 'incorrect argument', 'expected', 'syntax error',
|
|
'error', 'unparseable', 'invalid',
|
|
])
|
|
return {
|
|
'command': command,
|
|
'result': result.strip(),
|
|
'success': not is_error,
|
|
}
|
|
|
|
def get_server_status(self) -> Dict[str, Any]:
|
|
"""Get server state: players, time, difficulty."""
|
|
players_raw = rcon_send('list', self.host, self.port, self.password)
|
|
time_raw = rcon_send('time query daytime', self.host, self.port, self.password)
|
|
diff_raw = rcon_send('difficulty', self.host, self.port, self.password)
|
|
|
|
players = []
|
|
m = re.search(r'online:\s*(.*)', players_raw)
|
|
if m and m.group(1).strip():
|
|
players = [p.strip() for p in m.group(1).split(',') if p.strip()]
|
|
|
|
time_m = re.search(r'(\d+)', time_raw)
|
|
ticks = int(time_m.group(1)) if time_m else 0
|
|
|
|
diff_m = re.search(r'difficulty is (\w+)', diff_raw)
|
|
difficulty = diff_m.group(1) if diff_m else 'unknown'
|
|
|
|
return {
|
|
'players_online': players,
|
|
'player_count': len(players),
|
|
'time_ticks': ticks,
|
|
'difficulty': difficulty,
|
|
}
|
|
|
|
def get_player_info(self, player: str) -> Dict[str, Any]:
|
|
"""Get player position, health, gamemode."""
|
|
pos_raw = rcon_send(f'data get entity {player} Pos', self.host, self.port, self.password)
|
|
health_raw = rcon_send(f'data get entity {player} Health', self.host, self.port, self.password)
|
|
gm_raw = rcon_send(f'data get entity {player} playerGameType', self.host, self.port, self.password)
|
|
|
|
pos = None
|
|
pos_m = re.findall(r'(-?[\d.]+)d', pos_raw)
|
|
if pos_m and len(pos_m) >= 3:
|
|
pos = {'x': float(pos_m[0]), 'y': float(pos_m[1]), 'z': float(pos_m[2])}
|
|
|
|
health_m = re.search(r'([\d.]+)f', health_raw)
|
|
health = float(health_m.group(1)) if health_m else None
|
|
|
|
gm_m = re.search(r'data:\s*(\d+)', gm_raw)
|
|
gm_map = {0: 'survival', 1: 'creative', 2: 'adventure', 3: 'spectator'}
|
|
gamemode = gm_map.get(int(gm_m.group(1)), 'unknown') if gm_m else None
|
|
|
|
return {
|
|
'player': player,
|
|
'position': pos,
|
|
'health': health,
|
|
'gamemode': gamemode,
|
|
'online': pos is not None,
|
|
}
|