Spaces:
Running on Zero
Running on Zero
| """Compile a .bbproj block tree into a runnable nerimity.js bot, server-side. | |
| This is the whole security model. The client uploads *block structure* — opcodes, literal | |
| field values and nesting — never JavaScript. Every line of emitted code comes from | |
| catalog.json, which ships with the host and is not user-writable. An unknown opcode is | |
| rejected outright, and every literal is escaped through json.dumps or validated as a | |
| number, so a field value cannot break out of its expression and become code. | |
| Ported from the editor's original GDScript generator; catalog.json is generated from the | |
| same catalogue the editor renders from, so the two cannot drift. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import re | |
| from pathlib import Path | |
| from typing import Any | |
| CATALOG_PATH = Path(__file__).parent / "catalog.json" | |
| INDENT = " " | |
| # event name -> (handler signature, __ctx body) | |
| EVENT_CONTEXTS: dict[str, tuple[str, str]] = { | |
| "Ready": ("()", "{}"), | |
| "MessageCreate": ("(message)", | |
| "{ message, channel: message?.channel, user: message?.user, " | |
| "member: message?.member, server: message?.channel?.server }"), | |
| "MessageUpdate": ("(message)", | |
| "{ message, channel: message?.channel, user: message?.user, " | |
| "member: message?.member, server: message?.channel?.server }"), | |
| "MessageDelete": ("(payload)", | |
| "{ channel: client.channels?.cache?.get(payload?.channelId) }"), | |
| "MessageButtonClick": ("(button)", | |
| "{ button, message: button?.message, user: button?.user, channel: button?.channel }"), | |
| "MessageReactionAdded": ("(reaction, user)", | |
| "{ reaction, user, message: reaction?.message, channel: reaction?.channel }"), | |
| "MessageReactionRemoved": ("(reaction, user)", | |
| "{ reaction, user, message: reaction?.message, channel: reaction?.channel }"), | |
| "ServerMemberJoined": ("(member)", "{ member, user: member?.user, server: member?.server }"), | |
| "ServerMemberLeft": ("(member)", "{ member, user: member?.user, server: member?.server }"), | |
| } | |
| MAX_BLOCKS = 2000 # a project this large is a bug or an attack, not a bot | |
| MAX_DEPTH = 60 | |
| MAX_COMMAND_OPTIONS = 20 # keeps a command hat's preamble bounded | |
| # Command options are written the way people write lists: "count sides" or "count, sides". | |
| OPTION_SEPARATOR = re.compile(r"[\s,]+") | |
| class CompileError(Exception): | |
| """Raised for anything malformed; the message goes back to the editor verbatim.""" | |
| class Compiler: | |
| def __init__(self, catalog: dict | None = None) -> None: | |
| if catalog is None: | |
| catalog = json.loads(CATALOG_PATH.read_text()) | |
| self.blocks: dict[str, dict] = catalog["blocks"] | |
| self.variables: dict[str, str] = {} | |
| self.lists: dict[str, str] = {} | |
| self.warnings: list[str] = [] | |
| self._count = 0 | |
| # -- helpers ------------------------------------------------------------ | |
| def spec(self, opcode: str) -> dict: | |
| spec = self.blocks.get(opcode) | |
| if spec is None: | |
| raise CompileError(f"unknown block '{opcode}'") | |
| return spec | |
| def _ident(prefix: str, name: str) -> str: | |
| safe = re.sub(r"[^a-z0-9_]", "_", str(name).lower()) | |
| if not safe or safe[0].isdigit(): | |
| safe = "x" + safe | |
| return prefix + safe | |
| def _arg_def(self, spec: dict, name: str) -> dict | None: | |
| for a in spec["args"]: | |
| if a["name"] == name: | |
| return a | |
| return None | |
| def _command_options(node: dict) -> list[str]: | |
| """The option names declared on a command hat, in order and de-duplicated. | |
| Only the literal field is read. A reporter plugged into OPTIONS is ignored — option | |
| names have to be known while compiling, because each one becomes a variable. | |
| """ | |
| raw = str((node.get("fields") or {}).get("OPTIONS", "") or "").strip() | |
| if not raw: | |
| return [] | |
| out: list[str] = [] | |
| for name in OPTION_SEPARATOR.split(raw): | |
| if name and name not in out: | |
| out.append(name) | |
| return out[:MAX_COMMAND_OPTIONS] | |
| # -- literals ----------------------------------------------------------- | |
| def _literal(self, spec: dict, arg: dict, fields: dict) -> str: | |
| raw = fields.get(arg["name"], arg["default"]) | |
| raw = "" if raw is None else str(raw) | |
| kind = arg["type"] | |
| if kind == "variable": | |
| self.variables.setdefault(raw, self._ident("v_", raw)) | |
| return self.variables[raw] | |
| if kind == "list": | |
| self.lists.setdefault(raw, self._ident("l_", raw)) | |
| return self.lists[raw] | |
| if kind == "boolean": | |
| return "false" # empty hexagonal socket | |
| if kind == "dropdown": | |
| # Only values the catalogue offers; never echo an arbitrary string back into | |
| # the template, even though json.dumps would already neutralise it. | |
| options = arg.get("options") or [] | |
| if options and raw not in options: | |
| raw = str(arg["default"] or options[0]) | |
| return json.dumps(raw) | |
| if kind == "number": | |
| try: | |
| float(raw) | |
| return raw | |
| except ValueError: | |
| return json.dumps(raw) | |
| return json.dumps(raw) | |
| def _compile_arg(self, spec: dict, arg: dict, node: dict, depth: int) -> str: | |
| child = (node.get("inputs") or {}).get(arg["name"]) | |
| if child: | |
| return self._expression(child, depth + 1) | |
| return self._literal(spec, arg, node.get("fields") or {}) | |
| def _substitute(self, spec: dict, template: str, node: dict, depth: int) -> str: | |
| out = template | |
| for arg in spec["args"]: | |
| if arg["type"] == "substack": | |
| continue | |
| out = out.replace("{%s}" % arg["name"], self._compile_arg(spec, arg, node, depth)) | |
| return out | |
| # -- expressions -------------------------------------------------------- | |
| def _expression(self, node: dict, depth: int) -> str: | |
| self._guard_limits(depth) | |
| spec = self.spec(node.get("opcode", "")) | |
| js = self._substitute(spec, spec["js"], node, depth) | |
| if js.startswith("await "): | |
| return "(%s)" % js | |
| return js | |
| # -- statements --------------------------------------------------------- | |
| def _guard_limits(self, depth: int) -> None: | |
| self._count += 1 | |
| if self._count > MAX_BLOCKS: | |
| raise CompileError(f"project has more than {MAX_BLOCKS} blocks") | |
| if depth > MAX_DEPTH: | |
| raise CompileError(f"blocks nested deeper than {MAX_DEPTH}") | |
| def _stack(self, node: dict | None, depth: int) -> str: | |
| lines: list[str] = [] | |
| current = node | |
| seen = 0 | |
| while current: | |
| piece = self._statement(current, depth) | |
| if piece: | |
| lines.append(piece) | |
| current = current.get("next") | |
| seen += 1 | |
| if seen > MAX_BLOCKS: | |
| raise CompileError("block chain is impossibly long") | |
| return "\n".join(lines) | |
| def _statement(self, node: dict, depth: int) -> str: | |
| self._guard_limits(depth) | |
| spec = self.spec(node.get("opcode", "")) | |
| if not spec["js"].strip(): | |
| return "" | |
| pad = INDENT * depth | |
| template = self._substitute(spec, spec["js"], node, depth) | |
| substack_index = 0 | |
| for arg in spec["args"]: | |
| if arg["type"] != "substack": | |
| continue | |
| token = "{SUBSTACK}" if substack_index == 0 else "{SUBSTACK%d}" % (substack_index + 1) | |
| inner = self._stack((node.get("substacks") or {}).get(arg["name"]), depth + 1) | |
| template = template.replace(token, inner) | |
| substack_index += 1 | |
| if node.get("opcode") == "control_stop": | |
| if str((node.get("fields") or {}).get("WHAT", "this script")) == "all": | |
| template = "process.exit(0);" | |
| block_id = self._safe_id(node.get("id", "")) | |
| out = ['%s__bb.step("%s");' % (pad, block_id)] | |
| for line in template.split("\n"): | |
| if not line.strip(): | |
| continue | |
| out.append(line if line.startswith(INDENT * (depth + 1)) else pad + line) | |
| return "\n".join(out) | |
| def _safe_id(block_id: Any) -> str: | |
| # Block ids are echoed into a string literal for error attribution; keep them to a | |
| # character set that cannot terminate it. | |
| return re.sub(r"[^A-Za-z0-9_.-]", "", str(block_id))[:64] | |
| # -- top level ---------------------------------------------------------- | |
| def _collect_data(self, node: dict | None) -> None: | |
| if not node: | |
| return | |
| spec = self.spec(node.get("opcode", "")) | |
| fields = node.get("fields") or {} | |
| for arg in spec["args"]: | |
| value = str(fields.get(arg["name"], arg["default"]) or "") | |
| if arg["type"] == "variable": | |
| self.variables.setdefault(value, self._ident("v_", value)) | |
| elif arg["type"] == "list": | |
| self.lists.setdefault(value, self._ident("l_", value)) | |
| # A command hat's options are variables too, declared by naming them rather than by | |
| # sitting in a variable slot. | |
| if node.get("opcode") == "event_command": | |
| for option in self._command_options(node): | |
| self.variables.setdefault(option, self._ident("v_", option)) | |
| for child in (node.get("inputs") or {}).values(): | |
| self._collect_data(child) | |
| for child in (node.get("substacks") or {}).values(): | |
| self._collect_data(child) | |
| self._collect_data(node.get("next")) | |
| def _hat(self, node: dict) -> str: | |
| spec = self.spec(node["opcode"]) | |
| body = self._stack(node.get("next"), 1) | |
| if spec["event"] == "__broadcast": | |
| name_arg = self._arg_def(spec, "NAME") | |
| name = self._literal(spec, name_arg, node.get("fields") or {}) if name_arg else "''" | |
| lines = ["__bb.onBroadcast(%s, __bb.handler('broadcast', async () => {" % name, | |
| INDENT + "const __ctx = {};"] | |
| if body: | |
| lines.append(body) | |
| lines.append("}));") | |
| return "\n".join(lines) | |
| signature, ctx = EVENT_CONTEXTS.get(spec["event"], ("()", "{}")) | |
| lines = ["client.on(Events.%s, __bb.handler('%s', async %s => {" | |
| % (spec["event"], node["opcode"], signature), | |
| INDENT + "const __ctx = %s;" % ctx] | |
| if spec["event"] == "Ready": | |
| lines.append(INDENT + "__bb.ready(client);") | |
| else: | |
| lines.append(INDENT + "if (__ctx.user && client.user && " | |
| "__ctx.user.id === client.user.id) return;") | |
| if spec["guard"]: | |
| lines.append(INDENT + "if (!(%s)) return;" | |
| % self._substitute(spec, spec["guard"], node, 0)) | |
| # Options are read after the guard, so they only overwrite their variables when this | |
| # command is the one that actually ran. | |
| if node["opcode"] == "event_command": | |
| options = self._command_options(node) | |
| if options: | |
| lines.append(INDENT + "const __args = __ctx.message?.command?.args ?? [];") | |
| for index, option in enumerate(options): | |
| ident = self.variables.setdefault(option, self._ident("v_", option)) | |
| lines.append(INDENT + "%s = __bb.commandOption(__args, %d, %d);" | |
| % (ident, index, len(options))) | |
| if body: | |
| lines.append(body) | |
| lines.append("}));") | |
| return "\n".join(lines) | |
| def _command_registration(self, scripts: list[dict]) -> str: | |
| entries: list[str] = [] | |
| seen: set[str] = set() | |
| for node in scripts: | |
| if node.get("opcode") != "event_command": | |
| continue | |
| if (node.get("inputs") or {}).get("OPTIONS"): | |
| self.warnings.append( | |
| "A command hat uses a block for its options. Option names have to be " | |
| "plain text, because each one becomes a variable — that command will " | |
| "run with no options.") | |
| if (node.get("inputs") or {}).get("NAME"): | |
| self.warnings.append( | |
| "A command hat uses a block for its name, so it can't be registered " | |
| "automatically — type it as plain text to make it appear in Nerimity's " | |
| "command list.") | |
| continue | |
| fields = node.get("fields") or {} | |
| name = str(fields.get("NAME", "")).strip().lstrip("/") | |
| if not name or name in seen: | |
| continue | |
| seen.add(name) | |
| help_text = str(fields.get("HELP", "")).strip() or name | |
| # Nerimity shows `args` beside the command in its picker, so the option names | |
| # double as the usage hint people read before typing. | |
| usage = " ".join(self._command_options(node)) | |
| entries.append(" { name: %s, description: %s, args: %s }" | |
| % (json.dumps(name), json.dumps(help_text), json.dumps(usage))) | |
| if not entries: | |
| return "" | |
| return "\n".join([ | |
| "client.on(Events.Ready, __bb.handler('register_commands', async () => {", | |
| " await __bb.registerCommands(client, process.env.NERIMITY_TOKEN, [", | |
| ",\n".join(entries), | |
| " ]);", | |
| "}));", | |
| ]) | |
| def compile(self, project: dict, name: str = "bot") -> str: | |
| scripts = project.get("scripts") | |
| if not isinstance(scripts, list): | |
| raise CompileError("project has no scripts array") | |
| self.variables.clear() | |
| self.lists.clear() | |
| self.warnings.clear() | |
| self._count = 0 | |
| for node in scripts: | |
| if not isinstance(node, dict): | |
| raise CompileError("a script entry is not an object") | |
| self._collect_data(node) | |
| handlers: list[str] = [] | |
| orphans = 0 | |
| for node in scripts: | |
| if self.spec(node.get("opcode", ""))["shape"] != "hat": | |
| orphans += 1 | |
| continue | |
| handlers.append(self._hat(node)) | |
| if orphans: | |
| self.warnings.append( | |
| "%d loose block stack%s no event block on top, so %s never run." | |
| % (orphans, " has" if orphans == 1 else "s have", | |
| "it will" if orphans == 1 else "they will")) | |
| if not handlers: | |
| self.warnings.append( | |
| "This project has no event blocks, so the bot will connect and do nothing.") | |
| out = [ | |
| "'use strict';", | |
| "// Generated by the bbuilder host from block structure — do not edit.", | |
| "// project: %s" % re.sub(r"[\r\n]", " ", str(name))[:80], | |
| "", | |
| "const { Client, Events } = require('@nerimity/nerimity.js');", | |
| "const __bb = require('./bb-runtime.js');", | |
| "", | |
| "const client = new Client();", | |
| "", | |
| ] | |
| if self.variables or self.lists: | |
| out.append("// --- variables ---") | |
| # Two different names can sanitise to the same identifier ("my var" and "my-var", | |
| # or an option and a variable spelled with different capitals). They then share | |
| # one variable, which is the sane reading — but it must only be declared once. | |
| out += ["let %s = 0;" % v for v in dict.fromkeys(self.variables.values())] | |
| out += ["let %s = [];" % v for v in dict.fromkeys(self.lists.values())] | |
| out.append("") | |
| registration = self._command_registration(scripts) | |
| if registration: | |
| out.append("// --- slash command registration ---") | |
| out.append(registration) | |
| out.append("") | |
| for handler in handlers: | |
| out.append(handler) | |
| out.append("") | |
| out.append("client.login(process.env.NERIMITY_TOKEN);") | |
| out.append("") | |
| return "\n".join(out) | |
| def compile_project(project: dict, name: str = "bot") -> tuple[str, list[str]]: | |
| c = Compiler() | |
| return c.compile(project, name), list(c.warnings) |