#!/usr/bin/env python3 """Read-only public realm projection. Never proxy arbitrary routes or headers.""" import json import math import os import threading import time import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parents[1] / 'spectator' ALLIANCE = {'Human', 'Dwarf', 'Night Elf', 'Gnome', 'Draenei'} HORDE = {'Orc', 'Undead', 'Tauren', 'Troll', 'Blood Elf'} def text(value, limit=200): return str(value or '')[:limit] def number(value, default=0): return value if type(value) in (int, float) and math.isfinite(value) else default def project(state): """Explicit field allowlist: no settings, IDs, memories, whispers or host data.""" managed = [b for b in state.get('bots', []) if b.get('autonomy', {}).get('managed')] names = {b['name'] for b in managed} bots = [] parties = {} for b in managed: if b.get('status') != 'online': continue world = b.get('world') or {} race = text(b.get('race'), 40) team = world.get('team') faction = ('Alliance' if team == 0 else 'Horde' if team == 1 else 'Alliance' if race in ALLIANCE else 'Horde' if race in HORDE else 'Unknown') bots.append({'name': text(b.get('name'), 40), 'class': text(b.get('role'), 40), 'race': race, 'level': number(b.get('level')), 'zone': text(b.get('zone'), 100), 'faction': faction, 'guild': text(world.get('guildName'), 80)}) members = sorted({text(n, 40) for n in world.get('groupNames', []) if n in names}) if world.get('groupId') and len(members) > 1: parties[tuple(members)] = {'members': members, 'faction': faction} community = state.get('community') or {} feed = [] for e in community.get('feed', []): if e.get('status') != 'confirmed' or e.get('kind') not in {'conversation', 'relationship', 'party', 'leave', 'recruit', 'guild'}: continue # Only completed autonomous bot-to-bot exchanges. Human conversations, # whispers and synthetic opener prompts never enter this public feed. if e['kind'] == 'conversation' and not ( str(e.get('id', '')).startswith('social-') and str(e.get('id', '')).endswith(':reply') and len(e.get('names', [])) == 2 and all(n in names for n in e['names'])): continue feed.append({'at': text(e.get('at'), 40), 'kind': e['kind'], 'names': [text(n, 80) for n in e.get('names', [])[:5]], 'text': text(e.get('text'), 1200)}) guilds = [{'name': text(g.get('name'), 80), 'goal': text(g.get('goal'), 30), 'progress': number(g.get('progress')), 'target': number(g.get('target'), 20)} for g in community.get('guilds', [])[:100]] return {'rosterAt': number(state.get('aleRosterAt')), 'fetchedAt': int(time.time()), 'bots': sorted(bots, key=lambda b: (b['faction'], b['name'])), 'parties': list(parties.values()), 'guilds': guilds, 'feed': feed[-60:][::-1]} class Snapshot: def __init__(self): self.lock = threading.Lock() self.data = None self.checked = 0 self.failed = False def get(self): with self.lock: if time.monotonic() - self.checked > 10: try: # Fixed local endpoint. Viewer input cannot affect this request. opener = urllib.request.build_opener(NoRedirect(), urllib.request.ProxyHandler({})) with opener.open('http://127.0.0.1:8787/api/state', timeout=5) as response: raw = response.read(8 * 1024 * 1024 + 1) if len(raw) > 8 * 1024 * 1024: raise ValueError('State too large') self.data = project(json.loads(raw)) self.failed = False except Exception: self.failed = True self.checked = time.monotonic() if self.data is None: return None return {**self.data, 'stale': self.failed or time.time() - self.data['rosterAt'] > 120} class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None SNAPSHOT = Snapshot() ASSETS = {'/': ('index.html', 'text/html; charset=utf-8'), '/app.js': ('app.js', 'text/javascript; charset=utf-8'), '/style.css': ('style.css', 'text/css; charset=utf-8')} class Handler(BaseHTTPRequestHandler): server_version = 'AzerothAI-Spectator' sys_version = '' def setup(self): super().setup() self.connection.settimeout(10) def reply(self, status, body, content_type='application/json; charset=utf-8'): self.send_response(status) self.send_header('Content-Type', content_type) self.send_header('Content-Length', str(len(body))) self.send_header('Cache-Control', 'no-store') self.send_header('X-Content-Type-Options', 'nosniff') self.send_header('X-Frame-Options', 'DENY') self.send_header('Referrer-Policy', 'no-referrer') self.send_header('X-Robots-Tag', 'noindex, nofollow') self.send_header('Content-Security-Policy', "default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'none'") self.end_headers() if self.command != 'HEAD': self.wfile.write(body) def do_GET(self): path = self.path.split('?', 1)[0] if path in ASSETS: filename, content_type = ASSETS[path] self.reply(200, (ROOT / filename).read_bytes(), content_type) elif path == '/api/realm': data = SNAPSHOT.get() self.reply(200 if data else 503, json.dumps(data or {'error': 'Realm temporarily unavailable'}).encode()) elif path == '/health': self.reply(200, b'{"service":"spectator","readOnly":true}') elif path == '/robots.txt': self.reply(200, b'User-agent: *\nDisallow: /\n', 'text/plain; charset=utf-8') else: self.reply(404, b'{"error":"Not found"}') do_HEAD = do_GET def reject(self): self.close_connection = True self.reply(405, b'{"error":"This monitor is read-only"}') do_POST = do_PUT = do_PATCH = do_DELETE = do_OPTIONS = do_TRACE = do_CONNECT = reject def log_message(self, format, *args): pass class Server(ThreadingHTTPServer): daemon_threads = True slots = threading.BoundedSemaphore(24) def process_request(self, request, client_address): if not self.slots.acquire(blocking=False): self.shutdown_request(request) return try: super().process_request(request, client_address) except Exception: self.slots.release() raise def process_request_thread(self, request, client_address): try: super().process_request_thread(request, client_address) finally: self.slots.release() if __name__ == '__main__': Server(('127.0.0.1', int(os.environ.get('AZEROTHAI_SPECTATOR_PORT', '8790'))), Handler).serve_forever()