#!/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]} def public_state(state): """Admin-shaped data for the shared read-only UI, with an explicit allowlist.""" def pick(obj, keys): return {k: obj[k] for k in keys.split() if k in obj and isinstance(obj[k], (str, int, float, bool))} def recipe(r): return {**pick(r, 'id name description sociability aggression curiosity economy version'), 'tags': [text(t, 80) for t in r.get('tags', [])], 'activities': [a for a in r.get('activities', []) if a in ['quest','explore','trade','rest','socialize']]} projected = project(state) feed = [{**e, 'id': 'public-'+str(i), 'status': 'confirmed'} for i, e in enumerate(reversed(projected['feed']))] bots = [] for b in state.get('bots', []): if not b.get('autonomy', {}).get('managed'): continue bot = pick(b, 'id name role race level zone status tier mortal plotArmorHpFloor simulationBand homeId recipeId recipeVersion goal decisions') bot['personality'] = recipe(b.get('personality', {})) bot['routine'] = {**pick(b.get('routine', {}), 'wakeHour sleepHour'), 'preferredZones': b.get('routine', {}).get('preferredZones', [])} bot['autonomy'] = {'managed': True, **pick(b.get('autonomy', {}), 'observedAt lastIntentAt nextIntentAt')} world = b.get('world') or {} bot['world'] = {**pick(world, 'at team guildId guildName guildRank groupId inCombat x y mapId stationarySince'), 'groupNames': [n for n in world.get('groupNames', []) if any(x['name']==n for x in state['bots'] if x.get('autonomy',{}).get('managed'))], 'nearby': []} bot['relationships'] = [pick(r, 'targetId affinity note') for r in b.get('relationships', [])] # Show shared history, never personal whispers or operator-injected memories. bot['memories'] = [{'id': e['id'], 'text': e['text'], 'at': e['at'], 'source': 'observation', 'importance': 5} for e in feed if b['name'] in e['names']] if b.get('death'): bot['death'] = pick(b['death'], 'id at zone level cause killer source verified') if b.get('lastDecision'): bot['lastDecision'] = {'action': b['lastDecision']['action'], 'reason': 'Following the resident’s current activity.', 'goal': bot.get('goal','')} bots.append(bot) settings = state.get('settings') or {} return {'revision': state.get('revision',0), 'sequence': state.get('sequence',0), 'aleRosterAt': state.get('aleRosterAt',0), 'bots': bots, 'recipes': [recipe(r) for r in state.get('recipes',[])], 'homes': [{**pick(h,'id name zone type capacity storageSlots safety description'), 'ownerIds':h.get('ownerIds',[]), 'residentIds':h.get('residentIds',[])} for h in state.get('homes',[])], 'events': [{'id':e['id'],'seq':i,'at':e['at'],'kind':'relationship' if e['kind']=='relationship' else 'lifecycle','message':e['text'],'mode':'azerothcore'} for i,e in enumerate(feed)], 'settings': {**pick(settings,'connector provider model'), 'brains':pick(settings.get('brains',{}),'enabled residentModel heroModel backgroundUseLLM residentMemoryLimit heroMemoryLimit residentMaxTokens heroMaxTokens'), 'hardcore':pick(settings.get('hardcore',{}),'enabled requireVerifiedDeath memorialize'), 'population':pick(settings.get('population',{}),'defaultZoneTarget zoneMax serverMax foregroundMax backgroundMax')}, 'director':{**pick(state.get('director',{}),'experience zone brief'),'playerName':''}, 'population':{}, 'operations':{}, 'opsRetentionDays':state.get('opsRetentionDays',7), 'community':{'feed':feed,'evidence':[],'guilds':[{**g,'id':i} for i,g in enumerate(projected['guilds'])], 'partiesEnabled':state.get('community',{}).get('partiesEnabled',True), 'guildsEnabled':state.get('community',{}).get('guildsEnabled',True)}} def local_get(path): opener = urllib.request.build_opener(NoRedirect(), urllib.request.ProxyHandler({})) with opener.open('http://127.0.0.1:8787'+path, timeout=10) as response: return response.read(8 * 1024 * 1024) 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') state = json.loads(raw) self.data = {**project(state), "state": public_state(state)} 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' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; 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 ('/', '/visitor', '/visitor/'): try: self.reply(200, local_get('/visitor'), 'text/html; charset=utf-8') except Exception: self.reply(503, b'{"error":"Realm temporarily unavailable"}') elif path == '/api/state': data = SNAPSHOT.get() self.reply(200 if data else 503, json.dumps(data['state'] if data else {'error':'Realm unavailable'}).encode()) elif path == '/api/health': data = SNAPSHOT.get() self.reply(200, json.dumps({'ok':bool(data),'aleConfigured':True,'aleActiveRecently':bool(data and not data['stale']), 'llmConfigured':bool(data and data['state']['settings'].get('provider')!='deterministic')}).encode()) elif path == '/api/world-monitor': data = SNAPSHOT.get() at = data['rosterAt'] if data else 0 from datetime import datetime, timezone self.reply(200, json.dumps({'ok':bool(data),'ale':{'lastEventAt':datetime.fromtimestamp(at,timezone.utc).isoformat() if at else None,'activeRecently':bool(data and not data['stale'])},'events':[],'commands':{}}).encode()) elif path == '/api/realm-insights': try: # Fixed read-only endpoint: public aggregate facts and bot activity only. self.reply(200, local_get('/api/realm-insights')) except Exception: self.reply(503,b'{"error":"Realm insights unavailable"}') elif path == '/api/operations': try: health = json.loads(local_get('/api/operations')) samples = [{k:v for k,v in row.items() if k in ('host','at','cpuPercent','memoryTotalMB','memoryAvailableMB','diskFreeGB','databaseMB','logsMB','modelAvailable','runtimeService')} for row in health.get('samples',[])] self.reply(200, json.dumps({'samples':samples,'stuck':[{k:v for k,v in row.items() if k in ('name','zone','minutes')} for row in health.get('stuck',[])]}).encode()) except Exception: self.reply(503,b'{"error":"Health data unavailable"}') elif path.startswith(('/assets/', '/_vinext/', '/_next/')): # Built public assets only. No source maps, dotfiles or traversal. import mimetypes root = ROOT.parent / 'dist/client' asset = (root / path.lstrip('/')).resolve() if '%' in path or '..' in path or not asset.is_relative_to(root.resolve()) or asset.suffix not in ('.js','.css','.woff','.woff2','.png','.svg','.ico') or not asset.is_file(): self.reply(404,b'{"error":"Not found"}') else: self.reply(200,asset.read_bytes(),mimetypes.guess_type(str(asset))[0] or 'application/octet-stream') elif 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()