Verstack Local
f73167de85
Package Apple Silicon VM exports and cached runtime switching with local cabinet controls and conagent provisioning. Preserve experimental SPIKE 2 emulation. Improve preparation concurrency, Ghidra checkpoints, signature importing, sound indexing and client-side spectrograms. Include regression tests and validation notes.
423 lines
19 KiB
Python
423 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Bounded LAN pilot: isolated SPIKE sessions and same-origin JPEG/PCM media."""
|
|
import asyncio
|
|
from collections import deque
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import shutil
|
|
import signal
|
|
import struct
|
|
import time
|
|
from aiohttp import web
|
|
from PIL import Image, ImageDraw
|
|
import runtimes
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
RUNTIME = Path(os.environ.get('VERSTACK_EMULATOR_RUNTIME', '/runtime'))
|
|
SESSIONS = Path(os.environ.get('VERSTACK_EMULATOR_SESSIONS', '/sessions'))
|
|
MAX_SESSIONS = int(os.environ.get('VERSTACK_EMULATOR_MAX_SESSIONS', '2'))
|
|
TABLE = json.loads((ROOT / 'table.json').read_text())
|
|
EVENTS = {row[0] for row in TABLE['PLAYFIELD_SWITCHES']}
|
|
EVENTS.update(name for _, _, buttons, _ in TABLE['CONTROL_GROUPS'] for _, name in buttons)
|
|
EVENTS.update(('service_enter', 'service_back', 'service_up', 'service_down', 'coin_door_closed'))
|
|
|
|
|
|
async def rpc(session, operation, args=None):
|
|
async with asyncio.timeout(2):
|
|
reader, writer = await asyncio.open_unix_connection(str(session.directory / 'emulation/work/machine.sock'))
|
|
try:
|
|
writer.write(json.dumps({'v': 1, 'id': 'web', 'op': operation, 'args': args or {}}).encode() + b'\n')
|
|
await writer.drain()
|
|
reply = json.loads(await reader.readline())
|
|
if not reply.get('ok'):
|
|
raise ValueError(reply.get('error', {}).get('message', 'Machine rejected event'))
|
|
return reply['result']
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
|
|
class Session:
|
|
def __init__(self, mode, runtime=None):
|
|
self.id = secrets.token_hex(12)
|
|
self.token = secrets.token_urlsafe(32)
|
|
self.mode = mode
|
|
self.runtime = runtime
|
|
self.directory = SESSIONS / self.id
|
|
self.process = None
|
|
self.status = 'starting'
|
|
self.error = ''
|
|
self.created = time.monotonic()
|
|
self.last_seen = self.created
|
|
self.input_seen = self.last_seen
|
|
self.held = set()
|
|
self.input_lock = asyncio.Lock()
|
|
self.sequence = -1
|
|
self.events = deque(maxlen=20)
|
|
self.audio = deque(maxlen=5)
|
|
self.frame = b''
|
|
self.frame_number = 0
|
|
self.rendered = False
|
|
self.audio_bytes = 0
|
|
self.stream_connected = False
|
|
self.closed = False
|
|
self.task = None
|
|
|
|
async def start(self):
|
|
self.directory.mkdir(parents=True)
|
|
shutil.copytree(ROOT / 'vendor/emulation', self.directory / 'emulation')
|
|
if self.mode == 'diagnostic':
|
|
config = {
|
|
'socket': str(self.directory / 'emulation/work/machine.sock'),
|
|
'state_dir': str(self.directory / 'emulation/work/machine'),
|
|
'trace': {'enabled': False},
|
|
'i2c': {'eeprom': {'path': str(self.directory / 'eeprom.bin')}},
|
|
}
|
|
path = self.directory / 'machine.json'
|
|
path.write_text(json.dumps(config))
|
|
command = ['python3', str(self.directory / 'emulation/machine/service.py'), '--config', str(path)]
|
|
else:
|
|
for name in ('lib', 'usr', 'etc'):
|
|
(self.directory / name).symlink_to(Path(self.runtime['directory']) / self.runtime['system_root'] / name)
|
|
(self.directory / 'runtime.json').write_text(json.dumps(self.runtime))
|
|
games = self.directory / 'games'
|
|
games.mkdir()
|
|
title = games / 'title'
|
|
root = Path(self.runtime['directory'])
|
|
def link_file(source, destination):
|
|
Path(destination).symlink_to(source)
|
|
shutil.copytree(root / self.runtime['game_root'], title, copy_function=link_file)
|
|
if any(p.startswith('system/') for p in self.runtime['targets'].values()):
|
|
shutil.copytree(root / self.runtime['system_root'], title / 'system', copy_function=link_file)
|
|
command = ['python3', str(ROOT / 'runner.py'), str(self.directory), self.mode]
|
|
with (self.directory / 'console.log').open('wb') as log:
|
|
self.process = await asyncio.create_subprocess_exec(*command, stdout=log, stderr=log, start_new_session=True)
|
|
self.task = asyncio.create_task(self.capture())
|
|
|
|
async def release(self):
|
|
with contextlib.suppress(OSError, ValueError, asyncio.TimeoutError):
|
|
await rpc(self, 'switch.release_all')
|
|
self.held.clear()
|
|
|
|
async def stop(self):
|
|
if self.closed:
|
|
return
|
|
self.closed = True
|
|
async with self.input_lock:
|
|
await self.release()
|
|
if self.process:
|
|
# The entire process group belongs to this session, including sidecars.
|
|
with contextlib.suppress(ProcessLookupError):
|
|
os.killpg(self.process.pid, signal.SIGTERM)
|
|
with contextlib.suppress(asyncio.TimeoutError):
|
|
await asyncio.wait_for(self.process.wait(), 3)
|
|
with contextlib.suppress(ProcessLookupError):
|
|
os.killpg(self.process.pid, signal.SIGKILL)
|
|
await self.process.wait()
|
|
if self.task and self.task is not asyncio.current_task():
|
|
self.task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self.task
|
|
self.status = 'stopped'
|
|
shutil.rmtree(self.directory, ignore_errors=True)
|
|
|
|
def record(self):
|
|
return {'id': self.id, 'mode': self.mode, 'status': self.status, 'error': self.error,
|
|
'frames': self.frame_number, 'audio_bytes': self.audio_bytes,
|
|
'startup_seconds': int(time.monotonic() - self.created),
|
|
'startup_note': self.startup_note(),
|
|
'events': list(self.events), 'held': sorted(self.held),
|
|
'runtime': self.runtime['id'] if self.runtime else None,
|
|
'release': self.runtime['label'] if self.runtime else None}
|
|
|
|
def log_tail(self):
|
|
chunks = []
|
|
for label, path in (
|
|
('Launcher', self.directory / 'console.log'),
|
|
('Game', self.directory / 'emulation/work/capsule/writable/dump/debug_log.txt'),
|
|
):
|
|
if path.is_file():
|
|
with path.open('rb') as source:
|
|
source.seek(max(0, path.stat().st_size - 8192))
|
|
chunks.append(label + ':\n' + source.read().decode(errors='replace'))
|
|
return '\n'.join(chunks)
|
|
|
|
def startup_note(self):
|
|
if self.mode == 'diagnostic' or self.closed:
|
|
return ''
|
|
tail = self.log_tail()
|
|
if "Failed to open device '/dev/ttymxc1'" in tail:
|
|
return 'Cabinet startup is blocked: the SPIKE 2 node-bus device is not emulated yet.'
|
|
if self.rendered:
|
|
return ''
|
|
if 'failed to initialize bridge' in tail:
|
|
return 'Game startup failed at cabinet bridge initialization. No game frame has been rendered.'
|
|
if self.process and self.process.returncode is not None:
|
|
return 'The program exited before rendering a game frame.'
|
|
if time.monotonic() - self.created >= 60:
|
|
return 'No game frame after 60 seconds. Startup may be blocked; session logs are shown below.'
|
|
return 'Preparing runtime and starting the game; no game frame yet.'
|
|
|
|
def jpeg(self):
|
|
if self.mode == 'diagnostic':
|
|
image = Image.new('RGB', (960, 540), '#132739')
|
|
draw = ImageDraw.Draw(image)
|
|
draw.text((35, 40), 'VERSTACK STREAM / CONTROL DIAGNOSTIC', fill='white')
|
|
draw.text((35, 75), 'Synthetic picture and tone - no game running', fill='#ffcb70')
|
|
draw.text((35, 110), f'Private session {self.id} frame {self.frame_number}', fill='white')
|
|
draw.rectangle((int(time.monotonic() * 150) % 800, 170, int(time.monotonic() * 150) % 800 + 80, 230), fill='#59cdbb')
|
|
for index, event in enumerate(list(self.events)[-8:]):
|
|
draw.text((35, 280 + index * 25), str(event), fill='white')
|
|
else:
|
|
framebuffer = self.directory / 'emulation/work/capsule/framebuffer.xrgb'
|
|
if not framebuffer.exists():
|
|
return b''
|
|
data = framebuffer.read_bytes()
|
|
if len(data) != 1360 * 768 * 4:
|
|
return b''
|
|
image = Image.frombytes('RGB', (1360, 768), data, 'raw', 'BGRX')
|
|
if not self.rendered and image.getbbox() is None:
|
|
return b''
|
|
self.rendered = True
|
|
image = image.resize((960, 542))
|
|
encoded = io.BytesIO()
|
|
image.save(encoded, 'JPEG', quality=70)
|
|
return encoded.getvalue()
|
|
|
|
async def capture(self):
|
|
fd = None
|
|
tick = 0
|
|
try:
|
|
while not self.closed:
|
|
started = time.monotonic()
|
|
if self.process.returncode is not None:
|
|
self.status = 'completed' if self.process.returncode == 0 else 'failed'
|
|
self.error = '' if self.status == 'completed' else f'Emulator process exited ({self.process.returncode}). ' + self.log_tail().strip()[-1200:]
|
|
if self.status == 'completed':
|
|
self.frame = await asyncio.to_thread(self.jpeg)
|
|
self.frame_number += 1
|
|
log = self.directory / 'console.log'
|
|
with log.open('rb') as source:
|
|
source.seek(max(0, log.stat().st_size - 4096))
|
|
tail = source.read().decode(errors='replace')
|
|
if 'bwrap: Failed to make / slave: Permission denied' in tail:
|
|
self.error = 'Game launch blocked by the container mount policy (AppArmor). The stream/switch diagnostic works; changing game isolation requires administrator approval.'
|
|
elif 'bwrap: pivot_root: Operation not permitted' in tail:
|
|
self.error = 'Game launch blocked by seccomp: pivot_root is denied. An emulator-only syscall exception is prepared but requires administrator approval. The stream/switch diagnostic works.'
|
|
await self.release()
|
|
break
|
|
socket = self.directory / 'emulation/work/machine.sock'
|
|
if socket.exists():
|
|
self.status = 'diagnostic' if self.mode == 'diagnostic' else ('running' if self.rendered else 'initializing')
|
|
if tick % 4 == 0: # 12.5 fps portable diagnostic transport
|
|
self.frame = await asyncio.to_thread(self.jpeg)
|
|
if self.frame:
|
|
self.frame_number += 1
|
|
if self.mode == 'diagnostic':
|
|
pcm = b''.join(struct.pack('<hh', sample, sample) for sample in
|
|
(int(1800 * math.sin(2 * math.pi * 440 * (tick * 960 + i) / 48000)) for i in range(960)))
|
|
else:
|
|
fifo = self.directory / 'emulation/work/capsule/writable/run/spike3-emu/audio.pipe'
|
|
if fd is None and fifo.exists():
|
|
fd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK)
|
|
pcm = b''
|
|
if fd is not None:
|
|
with contextlib.suppress(BlockingIOError):
|
|
pcm = os.read(fd, 3840)
|
|
if pcm:
|
|
self.audio.append(pcm)
|
|
self.audio_bytes += len(pcm)
|
|
tick += 1
|
|
await asyncio.sleep(max(0, .02 - (time.monotonic() - started)))
|
|
finally:
|
|
if fd is not None:
|
|
os.close(fd)
|
|
|
|
|
|
@web.middleware
|
|
async def boundary(request, handler):
|
|
origin = request.headers.get('Origin')
|
|
if origin and origin != f'{request.scheme}://{request.host}':
|
|
raise web.HTTPForbidden(text='Cross-origin request rejected')
|
|
if request.method == 'POST' and request.headers.get('X-Verstack-Client') != '1':
|
|
raise web.HTTPForbidden(text='Missing client header')
|
|
try:
|
|
return await handler(request)
|
|
except (ValueError, OSError, asyncio.TimeoutError) as error:
|
|
return web.json_response({'error': str(error)}, status=400)
|
|
|
|
|
|
def owned(request):
|
|
session = request.app['sessions'].get(request.match_info['id'])
|
|
token = request.headers.get('Authorization', '').removeprefix('Bearer ')
|
|
if not session or not secrets.compare_digest(token, session.token):
|
|
raise web.HTTPNotFound(text='Session not found')
|
|
session.last_seen = time.monotonic()
|
|
return session
|
|
|
|
|
|
async def info(request):
|
|
available = runtimes.public_inventory(RUNTIME)
|
|
return web.json_response({'runtime_ready': any(r['ready'] for r in available),
|
|
'runtimes': available, 'max_sessions': MAX_SESSIONS,
|
|
'active_sessions': len(request.app['sessions']),
|
|
'table': TABLE, 'profile': 'Provisional donor mapping; select a validated machine profile for game-specific controls.'})
|
|
|
|
|
|
async def create(request):
|
|
body = await request.json()
|
|
if not isinstance(body, dict):
|
|
raise ValueError('JSON body must be an object')
|
|
mode = body.get('mode')
|
|
if mode not in ('diagnostic', 'game', 'boot-display', 'spike-menu'):
|
|
raise ValueError('Unknown launch mode')
|
|
runtime = None
|
|
if mode != 'diagnostic':
|
|
runtime = runtimes.select(RUNTIME, body.get('runtime', 'legacy'))
|
|
if mode not in runtime['targets']:
|
|
raise ValueError('This runtime does not contain the selected launch target')
|
|
sessions = request.app['sessions']
|
|
if len(sessions) >= MAX_SESSIONS:
|
|
raise web.HTTPConflict(text='Emulator capacity reached. Stop a session or wait for its idle timeout.')
|
|
session = Session(mode, runtime)
|
|
sessions[session.id] = session # Reserve capacity before yielding.
|
|
try:
|
|
await session.start()
|
|
except BaseException:
|
|
await session.stop()
|
|
sessions.pop(session.id, None)
|
|
raise
|
|
return web.json_response({**session.record(), 'token': session.token})
|
|
|
|
|
|
async def status(request):
|
|
session = owned(request)
|
|
result = session.record()
|
|
with contextlib.suppress(OSError, ValueError, asyncio.TimeoutError):
|
|
result['machine'] = await rpc(session, 'state')
|
|
return web.json_response(result)
|
|
|
|
|
|
async def event(request):
|
|
session = owned(request)
|
|
body = await request.json()
|
|
if not isinstance(body, dict):
|
|
raise ValueError('JSON body must be an object')
|
|
async with session.input_lock:
|
|
if body.get('release_all') is True:
|
|
await session.release()
|
|
return web.json_response({'ok': True})
|
|
if body.get('heartbeat') is True:
|
|
session.input_seen = time.monotonic()
|
|
return web.json_response({'ok': True})
|
|
sequence = body.get('sequence')
|
|
if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence <= session.sequence:
|
|
raise ValueError('Stale input sequence')
|
|
name = body.get('name')
|
|
if name not in EVENTS:
|
|
raise ValueError('Unknown switch')
|
|
args = {'name': name, 'duration_ms': 100}
|
|
if 'state' in body:
|
|
if not isinstance(body['state'], bool):
|
|
raise ValueError('Switch state must be boolean')
|
|
args['state'] = body['state']
|
|
result = await rpc(session, 'event', args)
|
|
session.sequence = sequence
|
|
session.input_seen = time.monotonic()
|
|
if body.get('state') is True:
|
|
session.held.add(name)
|
|
else:
|
|
session.held.discard(name)
|
|
session.events.append({'name': name, 'state': body.get('state', 'pulse')})
|
|
return web.json_response({'ok': True, 'result': result})
|
|
|
|
|
|
async def stop(request):
|
|
session = owned(request)
|
|
await session.stop()
|
|
request.app['sessions'].pop(session.id, None)
|
|
return web.json_response({'ok': True})
|
|
|
|
|
|
async def logs(request):
|
|
session = owned(request)
|
|
return web.Response(text=await asyncio.to_thread(session.log_tail))
|
|
|
|
|
|
async def media(request):
|
|
session = owned(request)
|
|
if session.stream_connected:
|
|
raise web.HTTPConflict(text='Session already has an active stream')
|
|
session.stream_connected = True
|
|
response = web.StreamResponse(headers={'Content-Type': 'application/octet-stream', 'Cache-Control': 'no-store', 'X-Accel-Buffering': 'no'})
|
|
previous = -1
|
|
try:
|
|
await response.prepare(request)
|
|
while not session.closed:
|
|
packets = []
|
|
if previous != session.frame_number and session.frame:
|
|
packets.append(b'V' + struct.pack('>I', len(session.frame)) + session.frame)
|
|
previous = session.frame_number
|
|
while session.audio:
|
|
pcm = session.audio.popleft()
|
|
packets.append(b'A' + struct.pack('>I', len(pcm)) + pcm)
|
|
if packets:
|
|
await asyncio.wait_for(response.write(b''.join(packets)), 2)
|
|
if session.status in ('failed', 'completed'):
|
|
break
|
|
await asyncio.sleep(.02)
|
|
except (ConnectionError, asyncio.TimeoutError):
|
|
pass
|
|
finally:
|
|
session.stream_connected = False
|
|
async with session.input_lock:
|
|
await session.release()
|
|
return response
|
|
|
|
|
|
async def lifetime(app):
|
|
SESSIONS.mkdir(parents=True, exist_ok=True)
|
|
async def reap():
|
|
while True:
|
|
await asyncio.sleep(1)
|
|
for session in list(app['sessions'].values()):
|
|
if time.monotonic() - session.input_seen > 3 and session.held:
|
|
async with session.input_lock:
|
|
await session.release()
|
|
if time.monotonic() - session.last_seen > 60:
|
|
await session.stop()
|
|
app['sessions'].pop(session.id, None)
|
|
task = asyncio.create_task(reap())
|
|
yield
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await task
|
|
await asyncio.gather(*(session.stop() for session in list(app['sessions'].values())))
|
|
|
|
|
|
def application():
|
|
app = web.Application(middlewares=[boundary], client_max_size=8192)
|
|
app['sessions'] = {}
|
|
app.cleanup_ctx.append(lifetime)
|
|
app.router.add_get('/emulator/info', info)
|
|
app.router.add_post('/emulator/sessions', create)
|
|
app.router.add_get('/emulator/sessions/{id}', status)
|
|
app.router.add_post('/emulator/sessions/{id}/event', event)
|
|
app.router.add_post('/emulator/sessions/{id}/stop', stop)
|
|
app.router.add_get('/emulator/sessions/{id}/media', media)
|
|
app.router.add_get('/emulator/sessions/{id}/logs', logs)
|
|
for url, name in (('/emulator/', 'index.html'), ('/emulator/client.js', 'client.js'), ('/emulator/table.svg', 'table.svg')):
|
|
async def static(request, name=name):
|
|
return web.FileResponse(ROOT / name)
|
|
app.router.add_get(url, static)
|
|
return app
|
|
|
|
|
|
if __name__ == '__main__':
|
|
web.run_app(application(), host='0.0.0.0', port=8095, access_log=None)
|