#!/usr/bin/env python3 """Hovermate Game Events SDK mock receiver (protocol v1), standard library only. Test a mod or a client without Hovermate. The mock writes its own connection.json into a temporary folder, accepts events over HTTP exactly like Hovermate (token, no Origin, size, batch and rate limits, the same validation and error codes), tails its inbox and any files you pass with --watch (JSONL event files and game logs with [HOVERMATE] lines), and prints every recognised event and every validation error. python mock-server.py # then set HOVERMATE_SDK_CONNECTION as printed and start the game python mock-server.py --watch "C:/Games/Cyberpunk 2077/bin/x64/plugins/cyber_engine_tweaks/mods/my_mod/hovermate-events.jsonl" Protocol: https://docs.hovermate.gg/sdk/protocol/ MIT License, Copyright (c) 2026 Paliverse Apps LLC. See LICENSE in the SDK. """ import argparse import base64 import datetime import json import os from pathlib import Path import re import secrets import signal import sys import tempfile import threading import time import unicodedata from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer ROUTE = '/hovermate/v1/events' MAXIMUM_BODY = 64 * 1024 DRAIN_LIMIT = 1024 * 1024 MAXIMUM_BATCH = 100 MAXIMUM_LINE = 4 * 1024 MAXIMUM_GAMES = 128 RATE_LIMIT = 20 MARKER = '[HOVERMATE] ' # The shared vocabulary of standard types (Docs/CLIPS.md in Hovermate). It only grows. VOCABULARY = frozenset([ 'kill', 'death', 'assist', 'multikill', 'roundWon', 'roundLost', 'matchWon', 'matchLost', 'mvp', 'goal', 'save', 'objective', 'mission', 'lap', 'bestLap', 'position', 'crash', 'incident', 'penalty', 'pit', 'finish', 'raceWon', 'landing', 'takeoff', 'fightWon', 'levelUp', 'bossKill', 'zone', 'achievement']) GAME_ID = re.compile(r'[a-z0-9._-]{3,64}') CUSTOM_TYPE = re.compile(r'[a-z0-9_.-]{1,32}') TAG = re.compile(r'[a-z0-9_-]{1,24}') INVALID_FILE_NAME = set('"<>|:*?\\/') | {chr(code) for code in range(32)} class Invalid(Exception): def __init__(self, code, detail): super().__init__(detail) self.code, self.detail = code, detail def _units(text): return len(text.encode('utf-16-le', 'surrogatepass')) // 2 def _plain(text, maximum): return _units(text) <= maximum and not any(unicodedata.category(c) == 'Cc' for c in text) def _text(parent, name): """A string field; an empty or whitespace-only string counts as missing (protocol section 3).""" value = parent.get(name) return value if isinstance(value, str) and value.strip() else None def parse_event(item): """The event as Hovermate reads it, or Invalid(code, detail). Checks run in Hovermate's order.""" if not isinstance(item, dict): raise Invalid('invalid_json', 'An event is a JSON object.') if type(item.get('v')) is not int or item['v'] != 1: raise Invalid('version', 'v must be 1.') game = item.get('game') if not isinstance(game, dict) or not isinstance(game.get('id'), str) or not GAME_ID.fullmatch(game['id']): raise Invalid('game', 'game.id must match [a-z0-9._-]{3,64}.') name = _text(game, 'name') if name is not None and not _plain(name, 64): raise Invalid('game', 'game.name is at most 64 characters.') exe = _text(game, 'exe') if exe is not None and (len(exe) > 128 or any(c in INVALID_FILE_NAME for c in exe)): raise Invalid('game', 'game.exe is a file name only.') kind = _text(item, 'type') if kind is None: raise Invalid('type', 'type is required.') event = {'gameId': game['id'], 'gameName': name.strip() if name is not None else None, 'exe': exe, 'type': kind} if kind == 'custom': custom_type, custom_label = _text(item, 'customType'), _text(item, 'customLabel') if (custom_type is None or not CUSTOM_TYPE.fullmatch(custom_type) or custom_label is None or not custom_label.strip() or not _plain(custom_label, 40)): raise Invalid('custom', 'custom events need customType [a-z0-9_.-]{1,32} and customLabel (at most 40 characters).') event.update(customType=custom_type, customLabel=custom_label.strip()) elif kind == 'hello': offers = [] if 'offers' in item: listed = item['offers'] if not isinstance(listed, list) or len(listed) > 250: raise Invalid('offers', 'offers is a list.') for offer in listed: if isinstance(offer, str) and offer in VOCABULARY: offers.append(offer) elif (isinstance(offer, dict) and isinstance(offer.get('customType'), str) and CUSTOM_TYPE.fullmatch(offer['customType']) and isinstance(offer.get('customLabel'), str) and offer['customLabel'].strip() and _plain(offer['customLabel'], 40)): offers.append({'customType': offer['customType'], 'customLabel': offer['customLabel'].strip()}) else: raise Invalid('offers', 'offers holds standard types or { customType, customLabel } objects.') event['offers'] = offers log = _text(item, 'log') if log is not None and not _plain(log, 260): raise Invalid('log', 'log is a file path of at most 260 characters.') event['log'] = log.strip() if log else None files = [] if 'files' in item: listed = item['files'] if (not isinstance(listed, list) or len(listed) > 16 or not all(isinstance(path, str) and (not path.strip() or _plain(path, 260)) for path in listed)): raise Invalid('files', 'files is a list of at most 16 file paths of at most 260 characters.') # A blank path is missing, like a blank log. files = [path.strip() for path in listed if path.strip()] event['files'] = files elif kind not in VOCABULARY: raise Invalid('type', 'Unknown standard type: ' + kind[:32] + '.') label = _text(item, 'label') if label is not None and not _plain(label, 80): raise Invalid('label', 'label is at most 80 characters.') tags = [] if 'tags' in item: listed = item['tags'] if not isinstance(listed, list) or len(listed) > 8: raise Invalid('tags', 'At most 8 tags.') for tag in listed: if not isinstance(tag, str) or not TAG.fullmatch(tag): raise Invalid('tags', 'Tags match [a-z0-9_-]{1,24}.') if tag not in tags: tags.append(tag) context = _text(item, 'context') if context is not None and not _plain(context, 64): raise Invalid('context', 'context is at most 64 characters.') event.update(label=label.strip() if label is not None else None, tags=tags, context=context.strip() if context is not None else None) stamp = _text(item, 'time') try: event['time'] = datetime.datetime.fromisoformat(stamp.replace('Z', '+00:00')) if stamp else None except ValueError: event['time'] = None # a time that does not parse is ignored return event def _depth(value, level=1): if isinstance(value, dict): return max([level] + [_depth(child, level + 1) for child in value.values()]) if isinstance(value, list): return max([level] + [_depth(child, level + 1) for child in value]) return level - 1 if level > 1 else 0 def _parse_json(data): def constant(name): raise ValueError(name) value = json.loads(data.decode('utf-8'), parse_constant=constant) if _depth(value) > 16: raise ValueError('too deep') return value class RateLimiter: """At most 20 events per second per game: a sliding one-second window.""" def __init__(self, limit=RATE_LIMIT): self.limit, self.windows, self.gate = limit, {}, threading.Lock() def try_take(self, game, count, now=None): now = time.monotonic() if now is None else now with self.gate: window = self.windows.setdefault(game, []) window[:] = [stamp for stamp in window if now - stamp < 1.0] if len(window) + count > self.limit: return False window.extend([now] * count) return True class Receiver: """The SDK side of Hovermate: games, rate limits, invalid counts, and what to do with valid events.""" def __init__(self, token, show=print, record=None): self.token, self.show, self.record = token, show, record self.games, self.invalid, self.limiter, self.gate = {}, {}, RateLimiter(), threading.Lock() self.accepted = 0 def _count(self, game_id): with self.gate: self.invalid[game_id] = self.invalid.get(game_id, 0) + 1 def _owner(self, item): game = item.get('game') if isinstance(item, dict) else None game_id = game.get('id') if isinstance(game, dict) else None return game_id if isinstance(game_id, str) and 3 <= len(game_id) <= 64 else None def known(self, game_id): with self.gate: return game_id in self.games def deliver(self, event, transport): """Creates or updates the game like Hovermate's SDK card; returns an error detail for a new nameless game.""" with self.gate: game = self.games.get(event['gameId']) if game is None: if not event['gameName']: return 'game.name is required in the first event or hello of a game.' if len(self.games) >= MAXIMUM_GAMES: return 'Too many SDK games.' game = self.games[event['gameId']] = {'name': event['gameName'], 'types': []} if event['gameName']: game['name'] = event['gameName'] types = [offer if isinstance(offer, str) else 'custom:' + offer['customType'] for offer in event.get('offers', [])] if event['type'] == 'hello' else \ ['custom:' + event['customType'] if event['type'] == 'custom' else event['type']] game['types'].extend(kind for kind in types if kind not in game['types']) self.accepted += 1 self._show_event(event, transport) return None def _show_event(self, event, transport): kind = event['type'] if kind == 'custom': kind = 'custom:%s "%s"' % (event['customType'], event['customLabel']) parts = ['%-5s' % transport, event['gameId'], kind] if event['type'] == 'hello': parts.append('offers=' + ','.join(o if isinstance(o, str) else 'custom:' + o['customType'] for o in event['offers'])) for path in ([event['log']] if event['log'] else []) + event['files']: parts.append('proposes "%s"' % path) if event.get('label'): parts.append('"%s"' % event['label']) if event.get('tags'): parts.append('tags=' + ','.join(event['tags'])) if event.get('context'): parts.append('context="%s"' % event['context']) self.show(' '.join(parts)) if self.record: data = {key: value for key, value in event.items() if value not in (None, [], '') and key != 'time'} self.record(dict(accepted=True, transport=transport, event=data)) def _refuse(self, transport, status, code, detail, index=None, game_id=None): where = '' if index is None else ' (event %d)' % index self.show('%-5s %d %s: %s%s' % (transport, status, code, detail, where)) if self.record: self.record(dict(accepted=False, transport=transport, status=status, error=code, detail=detail, index=index, gameId=game_id)) def handle(self, authorization, body): """POST /hovermate/v1/events after the transport checks: (status, JSON body).""" def error(status, code, detail, index=None, game_id=None): self._refuse('http', status, code, detail, index, game_id) reply = {'error': code, 'detail': detail} if index is not None: reply['index'] = index return status, json.dumps(reply, separators=(',', ':')) if not authorization or not authorization.startswith('Bearer ') or not secrets.compare_digest(authorization[7:].strip().encode(), self.token.encode()): return error(401, 'token', 'The bearer token is missing or wrong; read connection.json again.') if len(body) > MAXIMUM_BODY: return error(413, 'size', 'The body is larger than 64 KB.') try: root = _parse_json(body) except (ValueError, UnicodeDecodeError, RecursionError): return error(400, 'invalid_json', 'The body is not JSON.') batch = isinstance(root, dict) and 'events' in root items = (root['events'] if isinstance(root['events'], list) else None) if batch else [root] if items is None or not 0 < len(items) <= MAXIMUM_BATCH: return error(400, 'batch', 'events holds 1 to 100 events.') events = [] for index, item in enumerate(items): try: events.append(parse_event(item)) except Invalid as invalid: owner = self._owner(item) self._count(owner) return error(400, invalid.code, invalid.detail, index if batch else None, owner) mentioned = set() for index, event in enumerate(events): if event['gameId'] not in mentioned: mentioned.add(event['gameId']) if not event['gameName'] and not self.known(event['gameId']): self._count(event['gameId']) return error(400, 'game', 'game.name is required in the first event that mentions a new game.', index if batch else None, event['gameId']) counts = {} for event in events: if event['type'] != 'hello': counts[event['gameId']] = counts.get(event['gameId'], 0) + 1 for game_id, count in counts.items(): if not self.limiter.try_take(game_id, count): return error(429, 'rate', 'More than 20 events per second for ' + game_id + '.', None, game_id) accepted = 0 for event in events: refused = self.deliver(event, 'http') if refused: self._count(event['gameId']) return error(400, 'game', refused, None, event['gameId']) accepted += 1 return 202, json.dumps({'accepted': accepted}, separators=(',', ':')) def line(self, text, transport): """One event line of a file (inbox, mod-folder file or log).""" try: item = _parse_json(text.encode('utf-8')) except (ValueError, RecursionError): self._count(None) self._refuse(transport, 0, 'invalid_json', 'The line is not JSON.') return try: event = parse_event(item) except Invalid as invalid: owner = self._owner(item) self._count(owner) self._refuse(transport, 0, invalid.code, invalid.detail, game_id=owner) return if event['type'] != 'hello' and not self.limiter.try_take(event['gameId'], 1): self._refuse(transport, 0, 'rate', 'More than 20 events per second; the line is dropped.', game_id=event['gameId']) return refused = self.deliver(event, transport) if refused: self._count(event['gameId']) self._refuse(transport, 0, 'game', refused, game_id=event['gameId']) class Tail: """Follows one file like Hovermate: existing content is history, new lines count, truncation starts over. A write stamped more than 250 ms after the last read that leaves the file no longer than the offset rewrote it (a mod truncating its file at session start and writing the same bytes again) and starts over too.""" REWRITE_MARGIN_NS = 250_000_000 def __init__(self, path, from_start): self.path, self.offset, self.pending, self.log = Path(path), None, b'', None self.from_start = from_start self.written, self.read_at = 0, time.time_ns() if self.path.is_file() and not from_start: stat = self.path.stat() self.offset, self.written = stat.st_size, stat.st_mtime_ns def read(self): try: stat = self.path.stat() except OSError: if self.offset is not None: self.offset, self.pending = None, b'' return [] size = stat.st_size rewritten = (self.offset is not None and 0 < self.offset and size <= self.offset and stat.st_mtime_ns > self.written and stat.st_mtime_ns > self.read_at + self.REWRITE_MARGIN_NS) if self.offset is None or size < self.offset or rewritten: self.offset, self.pending = 0, b'' if size == self.offset: return [] with open(self.path, 'rb') as file: file.seek(self.offset) data = file.read(256 * 1024) self.written, self.read_at = os.fstat(file.fileno()).st_mtime_ns, time.time_ns() self.offset += len(data) data = self.pending + data lines = data.split(b'\n') self.pending = lines.pop() return [line.rstrip(b'\r').decode('utf-8', 'replace') for line in lines] class Files: """The inbox (*.jsonl, any game) and watched files, polled twice a second.""" def __init__(self, receiver, inbox, watched, follow_proposals): self.receiver, self.inbox, self.follow = receiver, Path(inbox), follow_proposals self.tails, self.watched, self.scanned = {}, [Path(path) for path in watched], False self.stop = threading.Event() def propose(self, path): if self.follow and Path(path) not in self.watched: self.receiver.show(' following proposed file "%s"' % path) self.watched.append(Path(os.path.expandvars(path))) def poll(self): wanted = [] if self.inbox.is_dir(): wanted += [(path, 'inbox', True) for path in sorted(self.inbox.glob('*.jsonl'))[:64]] wanted += [(path, 'file', False) for path in self.watched] for path, transport, inbox_file in wanted: tail = self.tails.get(path) if tail is None: appeared = self.scanned and (inbox_file or not path.exists()) tail = self.tails[path] = Tail(path, appeared or (not path.exists())) tail.transport = transport for text in tail.read(): marker = text.find(MARKER) if marker >= 0: json_text, tail.log = text[marker + len(MARKER):], True elif tail.log is not True and text.lstrip().startswith('{'): json_text, tail.log = text, False else: if tail.log is False and text.strip(): self.receiver._count(None) self.receiver._refuse(transport, 0, 'invalid_json', 'A line in an event file is not an event.') continue if len(json_text.encode('utf-8')) > MAXIMUM_LINE: self.receiver._count(None) self.receiver._refuse(transport, 0, 'size', 'The line is longer than 4 KB; dropped.') continue self.receiver.line(json_text, transport) self.scanned = True def run(self): while not self.stop.wait(0.5): try: self.poll() except OSError as failure: self.receiver.show(' file error: %s' % failure) def make_handler(receiver, port): class Handler(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' server_version = 'HovermateMock/1' def log_message(self, *args): # the mock prints events, not access logs pass def handle_expect_100(self): return True # like Hovermate: no "100 Continue", the client sends its body anyway def reply(self, status, body=''): data = body.encode('utf-8') self.send_response(status) if data: self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(data))) self.send_header('Connection', 'close') self.send_header('Cache-Control', 'no-store') self.end_headers() if data: self.wfile.write(data) self.close_connection = True def refuse(self, status, reason): receiver._refuse('http', status, reason, 'refused before the body was read') self.reply(status) def do_POST(self): headers = self.headers if self.path != ROUTE: return self.refuse(404, 'route') if 'Origin' in headers: return self.refuse(403, 'origin') host = headers.get('Host') if host is not None and host not in ('127.0.0.1:%d' % port, 'localhost:%d' % port): return self.refuse(403, 'host') if 'Transfer-Encoding' in headers: return self.refuse(411, 'length') kind = headers.get('Content-Type') if kind is not None and 'json' not in kind.lower(): return self.refuse(415, 'content_type') length = headers.get('Content-Length', '') if not length.isdigit() or int(length) <= 0: return self.refuse(411, 'length') if int(length) > MAXIMUM_BODY: if int(length) <= DRAIN_LIMIT: # Like Hovermate: a body up to 1 MB is read and dropped first (2 s at most), so the client sees the 413. self.connection.settimeout(2) try: left = int(length) while left > 0: chunk = self.rfile.read(min(left, 16384)) if not chunk: break left -= len(chunk) except OSError: pass return self.refuse(413, 'size') body = self.rfile.read(int(length)) status, reply = receiver.handle(headers.get('Authorization'), body) self.reply(status, reply) def do_GET(self): self.refuse(405, 'method') do_PUT = do_DELETE = do_PATCH = do_HEAD = do_OPTIONS = do_GET return Handler def main(): parser = argparse.ArgumentParser(description=__doc__.split('\n\n')[0]) parser.add_argument('--dir', type=Path, help='Folder for connection.json and the Inbox (default: a new temporary folder).') parser.add_argument('--port', type=int, default=0, help='HTTP port on 127.0.0.1 (default: a free port). This test address goes ' 'into the mock\'s connection.json; clients use the url from that file as written.') parser.add_argument('--watch', action='append', default=[], metavar='FILE', help='Also tail this file: a mod-folder hovermate-events.jsonl or a game log with [HOVERMATE] lines. Repeatable.') parser.add_argument('--follow-proposals', action='store_true', help='Tail the log and files a hello proposes (Hovermate asks the user first).') parser.add_argument('--record', type=Path, help='Append every accepted and refused event as one JSON line to this file.') parser.add_argument('--quiet', action='store_true', help='Print only the start-up lines.') args = parser.parse_args() folder = (args.dir or Path(tempfile.mkdtemp(prefix='hovermate-sdk-mock-'))).resolve() inbox = folder / 'Inbox' inbox.mkdir(parents=True, exist_ok=True) token = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b'=').decode() output_lock = threading.Lock() def show(message): if not args.quiet: with output_lock: print(datetime.datetime.now().strftime('[%H:%M:%S.%f')[:-3] + '] ' + message, flush=True) def record(entry): with output_lock, open(args.record, 'a', encoding='utf-8', newline='\n') as file: file.write(json.dumps(entry, ensure_ascii=False) + '\n') receiver = Receiver(token, show, record if args.record else None) files = Files(receiver, inbox, args.watch, args.follow_proposals) deliver = receiver.deliver def deliver_and_follow(event, transport): refused = deliver(event, transport) if not refused and event['type'] == 'hello': for path in ([event['log']] if event['log'] else []) + event['files']: files.propose(path) return refused receiver.deliver = deliver_and_follow server = ThreadingHTTPServer(('127.0.0.1', args.port), None) port = server.server_address[1] server.RequestHandlerClass = make_handler(receiver, port) connection = folder / 'connection.json' pending = connection.with_suffix('.writing') pending.write_text(json.dumps({'version': 1, 'url': 'http://127.0.0.1:%d%s' % (port, ROUTE), 'token': token, 'inbox': str(inbox)}, indent=2), encoding='utf-8') os.replace(pending, connection) files.poll() threading.Thread(target=files.run, daemon=True).start() print('Hovermate SDK mock listening on http://127.0.0.1:%d%s' % (port, ROUTE), flush=True) print('Connection file: %s' % connection, flush=True) print('Inbox: %s' % inbox, flush=True) print('PowerShell: $env:HOVERMATE_SDK_CONNECTION = "%s"' % connection, flush=True) print('cmd: set HOVERMATE_SDK_CONNECTION=%s' % connection, flush=True) print('Press Ctrl+C to stop.', flush=True) def stop(*_): raise KeyboardInterrupt signal.signal(signal.SIGTERM, stop) try: server.serve_forever(poll_interval=0.2) except KeyboardInterrupt: pass finally: files.stop.set() server.server_close() try: connection.unlink() except OSError: pass invalid = sum(receiver.invalid.values()) print('Stopped: %d accepted, %d invalid.' % (receiver.accepted, invalid), flush=True) if __name__ == '__main__': sys.exit(main())