"""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 @staticmethod 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 @staticmethod 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) @staticmethod 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)