320 lines
17 KiB
Python
Executable File
320 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Capture a real PNG through Paper and the Fabric worker using one online owner/spectator.
|
|
|
|
Does not start Minecraft/Paper, change configuration, or fabricate/alter images. By default,
|
|
uses the pose saved in game with /ai camera save test. Run with --delay 8 and return to the
|
|
Minecraft window, close menus, and keep the player still until the capture completes.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import binascii
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import struct
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
import zlib
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MAX_RESPONSE = 12 * 1024 * 1024
|
|
PNG_SIGNATURE = b'\x89PNG\r\n\x1a\n'
|
|
SAFE_METADATA = {
|
|
'status', 'captureId', 'capturedAt', 'dimension', 'x', 'y', 'z', 'eyeY', 'yaw', 'pitch',
|
|
'fov', 'readiness', 'serverRevisionVerified', 'loadedChunkRadius', 'stabilizationTicks',
|
|
'stabilizationFrames', 'afterOperationId', 'width', 'height', 'sourceWidth', 'sourceHeight',
|
|
'mimeType',
|
|
}
|
|
|
|
|
|
class TestFailure(Exception):
|
|
"""Messages are selected locally and never contain raw HTTP bodies or header values."""
|
|
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, request, response, code, message, headers, new_url):
|
|
raise TestFailure('Refusing an HTTP redirect from the local camera/Paper endpoint.')
|
|
|
|
|
|
def scalar(text, key):
|
|
matches = re.findall(r'^' + re.escape(key) + r':[ \t]*(.*?)[ \t]*$', text, re.M)
|
|
if len(matches) != 1:
|
|
raise TestFailure('Missing or duplicate top-level configuration field: ' + key)
|
|
raw = matches[0].strip()
|
|
if raw.startswith("'"):
|
|
match = re.fullmatch(r"'((?:[^']|'')*)'[ \t]*(?:#.*)?", raw)
|
|
if match:
|
|
return match.group(1).replace("''", "'")
|
|
elif raw.startswith('"'):
|
|
match = re.fullmatch(r'("(?:[^"\\]|\\.)*")[ \t]*(?:#.*)?', raw)
|
|
if match:
|
|
try:
|
|
value = json.loads(match.group(1))
|
|
if isinstance(value, str):
|
|
return value
|
|
except (ValueError, TypeError):
|
|
pass
|
|
else:
|
|
return re.split(r'[ \t]+#', raw, maxsplit=1)[0].strip()
|
|
raise TestFailure('Use a single Paper-generated scalar for configuration field: ' + key)
|
|
|
|
|
|
def identifier(value, label):
|
|
try:
|
|
parsed = str(uuid.UUID(value))
|
|
except (ValueError, TypeError, AttributeError):
|
|
raise TestFailure(label + ' must be configured as a UUID.') from None
|
|
return parsed
|
|
|
|
|
|
def port(value, label):
|
|
if not value.isdigit() or not 1 <= int(value) <= 65535:
|
|
raise TestFailure(label + ' must be a port in 1..65535.')
|
|
return int(value)
|
|
|
|
|
|
def instant(value, label):
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace('Z', '+00:00'))
|
|
if parsed.tzinfo is None:
|
|
raise ValueError()
|
|
return parsed.timestamp()
|
|
except (ValueError, TypeError, AttributeError):
|
|
raise TestFailure(label + ' is missing or invalid.') from None
|
|
|
|
|
|
def request_json(opener, url, token, body=None, timeout=10):
|
|
headers = {'Authorization': 'Bearer ' + token, 'Accept': 'application/json'}
|
|
if body is not None:
|
|
headers['Content-Type'] = 'application/json'
|
|
request = urllib.request.Request(url, headers=headers,
|
|
data=None if body is None else json.dumps(body, allow_nan=False).encode('utf-8'),
|
|
method='GET' if body is None else 'POST')
|
|
try:
|
|
with opener.open(request, timeout=timeout) as response:
|
|
raw = response.read(MAX_RESPONSE + 1)
|
|
if len(raw) > MAX_RESPONSE:
|
|
raise TestFailure('Backend response exceeds the 12 MiB limit.')
|
|
value = json.loads(raw)
|
|
if not isinstance(value, dict):
|
|
raise TestFailure('Backend returned an invalid JSON object.')
|
|
return value
|
|
except urllib.error.HTTPError as error:
|
|
# Do not relay the body, reason, request object, headers, or arbitrary exception text.
|
|
raise TestFailure('Local endpoint rejected the request (HTTP ' + str(error.code) + ').') from None
|
|
except (urllib.error.URLError, TimeoutError, ConnectionError, OSError):
|
|
raise TestFailure('Local endpoint is unavailable or timed out; check the running Paper/Fabric services.') from None
|
|
except (json.JSONDecodeError, UnicodeError):
|
|
raise TestFailure('Local endpoint returned invalid JSON.') from None
|
|
|
|
|
|
def rpc(opener, endpoint, token, scope, method, params=None, timeout=10):
|
|
reply = request_json(opener, endpoint + '/v1/rpc', token,
|
|
{'method': method, 'params': {**(params or {}), **scope}}, timeout=timeout)
|
|
if reply.get('ok') is not True or not isinstance(reply.get('result'), dict):
|
|
code = reply.get('error', {}).get('code', '') if isinstance(reply.get('error'), dict) else ''
|
|
safe_code = code if isinstance(code, str) and re.fullmatch(r'[a-z_]{1,64}', code) else 'request_failed'
|
|
raise TestFailure('Paper rejected the scoped request: ' + safe_code)
|
|
return reply['result']
|
|
|
|
|
|
def check_health(health, owner):
|
|
if health.get('status') != 'ok' or health.get('connected') is not True:
|
|
raise TestFailure('The Fabric camera is not connected to a world.')
|
|
if health.get('spectator') is not True:
|
|
raise TestFailure('The owner must already be in spectator mode; this script does not change game modes.')
|
|
if identifier(health.get('playerId'), 'Worker player ID') != owner:
|
|
raise TestFailure('The worker is using a different account from the configured owner.')
|
|
age = time.time() - instant(health.get('updatedAt'), 'Worker heartbeat')
|
|
if age < -5 or age > 5:
|
|
raise TestFailure('The worker heartbeat is stale; ensure Minecraft is ticking and the clocks agree.')
|
|
if health.get('busy') is True:
|
|
raise TestFailure('The camera is already busy; finish its current capture before running the test.')
|
|
|
|
|
|
def validate_pose(values):
|
|
x, y, z, yaw, pitch = values
|
|
if not all(math.isfinite(value) for value in values):
|
|
raise TestFailure('Pose values must all be finite numbers.')
|
|
if abs(x) > 29_999_984 or abs(z) > 29_999_984 or not -2048 <= y <= 2048:
|
|
raise TestFailure('Pose is outside the camera coordinate limits.')
|
|
if not -360 <= yaw <= 360 or not -90 <= pitch <= 90:
|
|
raise TestFailure('Yaw must be -360..360 and pitch -90..90.')
|
|
return dict(zip(('x', 'y', 'z', 'yaw', 'pitch'), values))
|
|
|
|
|
|
def inspect_png(raw):
|
|
"""Verify bytes received from the worker, without re-encoding or generating an image."""
|
|
if not raw.startswith(PNG_SIGNATURE) or len(raw) > 8 * 1024 * 1024:
|
|
raise TestFailure('Worker did not return a bounded PNG image.')
|
|
offset, dimensions, saw_pixels = 8, None, False
|
|
while offset + 12 <= len(raw):
|
|
length = struct.unpack_from('>I', raw, offset)[0]
|
|
kind = raw[offset + 4:offset + 8]
|
|
if length > len(raw) - offset - 12:
|
|
raise TestFailure('PNG chunk is truncated.')
|
|
data = raw[offset + 8:offset + 8 + length]
|
|
expected = struct.unpack_from('>I', raw, offset + 8 + length)[0]
|
|
if zlib.crc32(kind + data) & 0xffffffff != expected:
|
|
raise TestFailure('PNG checksum does not match the received data.')
|
|
if offset == 8:
|
|
if kind != b'IHDR' or length != 13:
|
|
raise TestFailure('PNG has no valid IHDR header.')
|
|
dimensions = struct.unpack_from('>II', data)
|
|
if not 1 <= dimensions[0] <= 1920 or not 1 <= dimensions[1] <= 1080:
|
|
raise TestFailure('PNG dimensions exceed the output limits.')
|
|
if kind == b'IDAT':
|
|
saw_pixels = True
|
|
offset += length + 12
|
|
if kind == b'IEND':
|
|
if length != 0 or offset != len(raw) or not saw_pixels:
|
|
raise TestFailure('PNG is incomplete or contains trailing data.')
|
|
return dimensions
|
|
raise TestFailure('PNG does not contain a complete image.')
|
|
|
|
|
|
def write_private(path, data):
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, 'wb') as out:
|
|
out.write(data)
|
|
out.flush()
|
|
os.fsync(out.fileno())
|
|
|
|
|
|
def run(args):
|
|
try:
|
|
text = args.config.read_text(encoding='utf-8')
|
|
except (OSError, UnicodeError):
|
|
raise TestFailure('Cannot read the private Paper config; start/configure the plugin first.') from None
|
|
owner = identifier(scalar(text, 'owner-uuid'), 'Owner UUID')
|
|
camera_player = identifier(scalar(text, 'camera-player-uuid'), 'Camera player UUID')
|
|
if owner != camera_player:
|
|
raise TestFailure('This single-client test requires camera-player-uuid to equal owner-uuid.')
|
|
agent_token, camera_token = scalar(text, 'agent-token'), scalar(text, 'camera-token')
|
|
if any(not re.fullmatch(r'[A-Za-z0-9._~-]{32,512}', value) for value in (agent_token, camera_token)) or agent_token == camera_token:
|
|
raise TestFailure('Agent and camera tokens must be distinct safe values; contents are not displayed.')
|
|
project = scalar(text, 'project-id')
|
|
if not project or len(project) > 256:
|
|
raise TestFailure('A valid project-id is required.')
|
|
endpoint = 'http://127.0.0.1:' + str(port(scalar(text, 'http-port'), 'http-port'))
|
|
worker = 'http://127.0.0.1:' + str(port(scalar(text, 'camera-port'), 'camera-port'))
|
|
scope = {'project_id': project, 'player_id': owner}
|
|
# Never let HTTP_PROXY/HTTPS_PROXY forward private capability headers away from loopback.
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
|
|
health = request_json(opener, worker + '/health', camera_token)
|
|
check_health(health, owner)
|
|
# This authenticates the real in-game owner and confirms the configured Paper server is reachable.
|
|
context = rpc(opener, endpoint, agent_token, scope, 'project_context')
|
|
if context.get('project_id') != project:
|
|
raise TestFailure('Paper returned a different project scope.')
|
|
if args.pose:
|
|
params = {'pose': validate_pose(args.pose)}
|
|
if args.fov is not None:
|
|
params['pose']['fov'] = args.fov
|
|
pose_label = 'explicit pose'
|
|
else:
|
|
cameras = rpc(opener, endpoint, agent_token, scope, 'camera_list')
|
|
if not any(isinstance(camera, dict) and camera.get('camera_id') == args.camera_id for camera in cameras.get('cameras', [])):
|
|
raise TestFailure('Saved camera was not found. In Minecraft run /ai camera save test, or pass --pose X Y Z YAW PITCH.')
|
|
params = {'camera_id': args.camera_id}
|
|
pose_label = 'saved camera'
|
|
if args.after_operation_id:
|
|
params['after_operation_id'] = identifier(args.after_operation_id, 'Operation ID')
|
|
print('Ready: online owner/spectator and scoped Paper route verified; using ' + pose_label + '.', flush=True)
|
|
print('Return to Minecraft, close chat/menus, keep the window rendering, and do not move the mouse or player. Capture starts in '
|
|
+ str(args.delay) + ' seconds.', flush=True)
|
|
time.sleep(args.delay)
|
|
check_health(request_json(opener, worker + '/health', camera_token), owner)
|
|
started = time.time()
|
|
deadline = time.monotonic() + args.timeout
|
|
result = rpc(opener, endpoint, agent_token, scope, 'camera_capture', params, timeout=min(35, args.timeout))
|
|
capture_id = identifier(result.get('captureId'), 'Capture ID')
|
|
print('Paper accepted the capture; waiting for the real rendered frame.', flush=True)
|
|
while result.get('status') == 'pending':
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TestFailure('Capture did not complete within the test deadline; no image was saved.')
|
|
time.sleep(min(0.5, remaining))
|
|
result = rpc(opener, endpoint, agent_token, scope, 'camera_capture', {'capture_id': capture_id}, timeout=min(10, max(0.5, remaining)))
|
|
if result.get('status') != 'completed':
|
|
code = result.get('error', '')
|
|
safe_code = code if isinstance(code, str) and re.fullmatch(r'[a-z_]{1,64}', code) else 'capture_failed'
|
|
raise TestFailure('The camera returned ' + safe_code + '; no image was saved.')
|
|
if result.get('captureId') != capture_id or result.get('mimeType') != 'image/png':
|
|
raise TestFailure('Completed capture identity or MIME type did not match.')
|
|
captured_at = instant(result.get('capturedAt'), 'Capture timestamp')
|
|
if captured_at < started - 1 or captured_at > time.time() + 5:
|
|
raise TestFailure('The frame timestamp does not correspond to this capture request.')
|
|
encoded = result.get('imageBase64')
|
|
if not isinstance(encoded, str) or len(encoded) > 12_000_000:
|
|
raise TestFailure('Completed capture did not include a bounded image payload.')
|
|
try:
|
|
raw = base64.b64decode(encoded, validate=True)
|
|
except (ValueError, binascii.Error):
|
|
raise TestFailure('Completed capture contained invalid image encoding.') from None
|
|
width, height = inspect_png(raw)
|
|
if result.get('width') != width or result.get('height') != height:
|
|
raise TestFailure('PNG dimensions disagree with the capture metadata.')
|
|
metadata = {key: value for key, value in result.items() if key in SAFE_METADATA}
|
|
# Include only known primitive fields; a backend must not smuggle an arbitrary nested payload here.
|
|
if any(not isinstance(value, (str, int, float, bool, type(None))) for value in metadata.values()):
|
|
raise TestFailure('Capture metadata contains unexpected structured values.')
|
|
metadata.update(imageSha256=hashlib.sha256(raw).hexdigest(), imageBytes=len(raw), elapsedSeconds=round(time.time() - started, 3),
|
|
transport='authenticated Paper HTTP camera_capture', testMode='one owner/spectator client')
|
|
safe_json = json.dumps(metadata, ensure_ascii=False, allow_nan=False, indent=2).encode('utf-8') + b'\n'
|
|
if agent_token.encode() in safe_json or camera_token.encode() in safe_json:
|
|
raise TestFailure('Refusing to save metadata containing a component secret.')
|
|
directory = ROOT / '.runtime/camera-test'
|
|
directory.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
prefix = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') + '-' + capture_id[:8]
|
|
image_path, metadata_path = directory / (prefix + '.png'), directory / (prefix + '.json')
|
|
write_private(image_path, raw)
|
|
write_private(metadata_path, safe_json)
|
|
print('Saved real PNG: ' + str(image_path), flush=True)
|
|
print('Saved sanitized metadata: ' + str(metadata_path), flush=True)
|
|
print('Frame size: ' + str(width) + 'x' + str(height) + '. Server revision verified: '
|
|
+ str(result.get('serverRevisionVerified') is True).lower() + '.', flush=True)
|
|
return 0
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
parser.add_argument('--config', type=Path, default=ROOT / '.runtime/server/plugins/MinecraftBuilderMCP/config.yml')
|
|
source = parser.add_mutually_exclusive_group()
|
|
source.add_argument('--camera-id', default='test', help='Saved in-game camera name; default: test')
|
|
source.add_argument('--pose', type=float, nargs=5, metavar=('X', 'Y', 'Z', 'YAW', 'PITCH'), help='Explicit feet position and viewing angles')
|
|
parser.add_argument('--fov', type=int, help='Field of view for an explicit --pose (30..110 degrees)')
|
|
parser.add_argument('--after-operation-id', help='Optional completed operation UUID for Paper validation and correlation')
|
|
parser.add_argument('--delay', type=int, default=8, help='Seconds to return to the Minecraft window before capture (0..60)')
|
|
parser.add_argument('--timeout', type=int, default=45, help='Total capture/poll budget in seconds (20..120)')
|
|
args = parser.parse_args()
|
|
if args.fov is not None and (not args.pose or not 30 <= args.fov <= 110):
|
|
parser.error('--fov requires --pose and must be 30..110')
|
|
if not 0 <= args.delay <= 60 or not 20 <= args.timeout <= 120:
|
|
parser.error('--delay must be 0..60 and --timeout must be 20..120')
|
|
if args.camera_id and (not args.camera_id.strip() or len(args.camera_id) > 64):
|
|
parser.error('--camera-id must contain 1..64 characters')
|
|
try:
|
|
return run(args)
|
|
except TestFailure as error:
|
|
print('Camera test failed: ' + str(error), file=sys.stderr)
|
|
return 1
|
|
except KeyboardInterrupt:
|
|
print('Camera test interrupted; the worker will expire its current request.', file=sys.stderr)
|
|
return 130
|
|
except Exception:
|
|
# Unexpected errors can include HTTP header details; never print a traceback with private state.
|
|
print('Camera test failed unexpectedly; no private configuration or raw exception is printed.', file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|