diff --git a/surfaces/blender/brokkr_bridge/__init__.py b/surfaces/blender/brokkr_bridge/__init__.py index 4720cde..ee7862f 100644 --- a/surfaces/blender/brokkr_bridge/__init__.py +++ b/surfaces/blender/brokkr_bridge/__init__.py @@ -19,6 +19,7 @@ from .blender_target import ( TOOL_KIND, BrokkrBlenderTarget, ) +from .server_guard import BindRefused _target = None @@ -86,6 +87,19 @@ class BrokkrPreferences(bpy.types.AddonPreferences): description="Port for the Blender CultMesh server; 0 asks the OS for a free port", ) + allow_non_loopback: bpy.props.BoolProperty( + name="Allow Non-Loopback Bind", + default=False, + update=lambda self, context: target().enforce_bind_policy(self.allow_non_loopback), + description="Let the server bind an address other than loopback (for example 0.0.0.0), which exposes it beyond this machine. Set by hand; Start Server refuses a non-loopback Serve Host without it", + ) + + require_session_token: bpy.props.BoolProperty( + name="Require Session Token", + default=False, + description="Optional. Generate a random token at Start Server, show it in the panel, and reject any document put or delete that lacks it in the sessionToken field. Off by default; the token is never saved", + ) + max_snapshot_documents: bpy.props.IntProperty( name="Max Snapshot Documents", default=1000, @@ -260,6 +274,8 @@ class BrokkrPreferences(bpy.types.AddonPreferences): layout.prop(self, "auto_drain_commands") layout.prop(self, "serve_host") layout.prop(self, "serve_port") + layout.prop(self, "allow_non_loopback") + layout.prop(self, "require_session_token") layout.prop(self, "max_snapshot_documents") layout.prop(self, "max_snapshot_bytes") layout.prop(self, "sync_session_id") @@ -307,6 +323,17 @@ class BROKKR_PT_status(bpy.types.Panel): layout.label(text=f"Node: {adapter.resolve_cache_path(prefs.cultmesh_cache_path)}") server = adapter.server_status() layout.label(text=f"Server: {server['endpoint'] if server['running'] else 'stopped'}") + if server["running"]: + layout.label(text=f"Bind: {server['host']}:{server['port']}") + layout.label(text=f"cultmesh_py loaded from: {server['cultmeshModule']}") + if not server["loopback"]: + warning = layout.box() + warning.alert = True + warning.label(text="WARNING: bound beyond loopback; reachable from the network", icon="ERROR") + if server["tokenRequired"]: + layout.label(text="Session token (send as sessionToken on every put):") + layout.label(text=adapter.session_token) + layout.operator("brokkr.copy_session_token", icon="COPYDOWN") row = layout.row(align=True) row.operator("brokkr.capture_snapshot", icon="FILE_REFRESH") @@ -425,18 +452,40 @@ class BROKKR_OT_start_server(bpy.types.Operator): def execute(self, context): prefs = context.preferences.addons[__name__].preferences - status = target().start_server( - prefs.cultmesh_cache_path, - prefs.cultlib_py_src, - prefs.serve_host, - prefs.serve_port, - prefs.max_snapshot_documents, - prefs.max_snapshot_bytes, - ) + try: + status = target().start_server( + prefs.cultmesh_cache_path, + prefs.cultlib_py_src, + prefs.serve_host, + prefs.serve_port, + prefs.max_snapshot_documents, + prefs.max_snapshot_bytes, + prefs.debug_mirror_root, + prefs.allow_non_loopback, + prefs.require_session_token, + ) + except BindRefused as exc: + self.report({"ERROR"}, str(exc)) + return {"CANCELLED"} self.report({"INFO"}, f"Brokkr Blender server: {status['endpoint']}") return {"FINISHED"} +class BROKKR_OT_copy_session_token(bpy.types.Operator): + bl_idname = "brokkr.copy_session_token" + bl_label = "Copy Session Token" + bl_description = "Copy the running server's session token to the clipboard" + + def execute(self, context): + token = target().session_token + if not token: + self.report({"WARNING"}, "No session token: the server is stopped or the token is off.") + return {"CANCELLED"} + context.window_manager.clipboard = token + self.report({"INFO"}, "Brokkr session token copied.") + return {"FINISHED"} + + class BROKKR_OT_stop_server(bpy.types.Operator): bl_idname = "brokkr.stop_server" bl_label = "Stop Server" @@ -588,6 +637,7 @@ classes = ( BROKKR_OT_drain_commands, BROKKR_OT_start_server, BROKKR_OT_stop_server, + BROKKR_OT_copy_session_token, BROKKR_OT_publish_object_sync, BROKKR_OT_publish_timeline_sync, BROKKR_OT_refresh_sync_receipt, diff --git a/surfaces/blender/brokkr_bridge/blender_target.py b/surfaces/blender/brokkr_bridge/blender_target.py index 2bfe3c4..88c9f20 100644 --- a/surfaces/blender/brokkr_bridge/blender_target.py +++ b/surfaces/blender/brokkr_bridge/blender_target.py @@ -9,6 +9,8 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable +from .server_guard import BindRefused, SessionGuard, check_bind, guarded_server_class, is_loopback_host, purge_command_intents + PROVIDER_ID = "brokkr.blender_editor" TOOL_KIND = "blender-editor" DEFAULT_BROKER_URI = "cultmesh://brokkr" @@ -60,6 +62,9 @@ class BrokkrBlenderTarget: self._documents: dict[str, Any] | None = None self._server_key: tuple[str, str, str, str, int, int, int] | None = None self._server: Any | None = None + self._guard: SessionGuard | None = None + self._cultmesh_module: str = "" + self._debug_mirror_root: str = "" def resolve_cache_path(self, cache_path: str) -> str: return self.bpy.path.abspath(cache_path or DEFAULT_CULTMESH_CACHE_PATH) @@ -291,9 +296,17 @@ class BrokkrBlenderTarget: port: int, max_snapshot_documents: int, max_snapshot_bytes: int, + debug_mirror_root: str = "", + allow_non_loopback: bool = False, + require_token: bool = False, ) -> dict[str, Any]: + try: + resolved_host = check_bind(host, allow_non_loopback) + except BindRefused: + self.enforce_bind_policy(allow_non_loopback) + raise + self.enforce_bind_policy(allow_non_loopback) node, documents = self._open_node(cache_path, cultlib_py_src) - resolved_host = host or "127.0.0.1" server_key = ( self.resolve_cache_path(cache_path), str(Path(cultlib_py_src or DEFAULT_CULTLIB_PY_SRC)), @@ -302,16 +315,25 @@ class BrokkrBlenderTarget: int(port), int(max_snapshot_documents), int(max_snapshot_bytes), + bool(require_token), ) if self._server is not None and self._server_key == server_key: return self.server_status() self.stop_server() + self._cultmesh_module = "" + self._debug_mirror_root = self.resolve_debug_mirror_root(debug_mirror_root) if debug_mirror_root else "" + purge_command_intents(node, documents["command_intent"], COMMAND_INTENT_SCHEMA, self._debug_mirror_root) cultmesh, _cultcache = _load_cultmesh(cultlib_py_src) + self._cultmesh_module = str(getattr(cultmesh, "__file__", "") or "") verse_catalog = cultmesh.CultMesh.create_verse_catalog() peer_catalog = cultmesh.CultMesh.create_peer_catalog() - server = cultmesh.CultMesh.serve_node( - node, + server_class = cultmesh.CultMeshLocalServer + guard = SessionGuard() if require_token else None + if guard is not None: + server_class = guarded_server_class(server_class, guard) + server = server_class( + node=node, verse_catalog=verse_catalog, peer_catalog=peer_catalog, host=resolved_host, @@ -319,7 +341,7 @@ class BrokkrBlenderTarget: display_name="Brokkr Blender Editor", max_snapshot_documents=int(max_snapshot_documents), max_snapshot_bytes=int(max_snapshot_bytes), - ) + ).start() endpoint = f"cultnet://{resolved_host}:{server.port}" verse_catalog.upsert(cultmesh.CultMeshVerseDescriptor( @@ -345,13 +367,36 @@ class BrokkrBlenderTarget: self._server = server self._server_key = server_key + self._guard = guard return self.server_status() + def enforce_bind_policy(self, allow_non_loopback: bool) -> None: + """Stop a running server that is bound beyond loopback when the allow is not set.""" + if self._server is not None and not allow_non_loopback and not is_loopback_host(self._server.host): + self.stop_server() + def stop_server(self) -> None: - if self._server is not None: - self._server.stop() - self._server = None - self._server_key = None + was_running = self._server is not None + try: + if was_running: + self._server.stop() + finally: + try: + if self._guard is not None: + self._guard.revoke() + finally: + try: + if was_running and self._node is not None: + purge_command_intents(self._node, self._documents["command_intent"], COMMAND_INTENT_SCHEMA, self._debug_mirror_root) + finally: + self._server = None + self._server_key = None + self._guard = None + + @property + def session_token(self) -> str: + """The running server's token, or "" when the server is stopped or the token is off.""" + return self._guard.token if self._guard is not None else "" def server_status(self) -> dict[str, Any]: if self._server is None: @@ -360,12 +405,18 @@ class BrokkrBlenderTarget: "endpoint": "", "host": "", "port": 0, + "loopback": True, + "tokenRequired": False, + "cultmeshModule": "", } return { "running": True, "endpoint": f"cultnet://{self._server.host}:{self._server.port}", "host": self._server.host, "port": self._server.port, + "loopback": is_loopback_host(self._server.host), + "tokenRequired": self._guard is not None, + "cultmeshModule": self._cultmesh_module, } def capture_snapshot(self, context: Any) -> dict[str, Any]: diff --git a/surfaces/blender/brokkr_bridge/server_guard.py b/surfaces/blender/brokkr_bridge/server_guard.py new file mode 100644 index 0000000..0e9d111 --- /dev/null +++ b/surfaces/blender/brokkr_bridge/server_guard.py @@ -0,0 +1,107 @@ +"""Server hardening for the Blender add-on's CultMesh server. + +Stdlib only, no bpy: everything here is unit-testable without Blender. + +Always on: a non-loopback bind needs an explicit allow (check_bind), and command +intents queued before a start or stop are purged (purge_command_intents). +Opt-in: a per-session token, checked on every document put and delete when the +user turns it on in the add-on preferences (SessionGuard, guarded_server_class). +""" +from __future__ import annotations + +import hmac +import ipaddress +import secrets +from pathlib import Path +from typing import Any + +PUT_SCHEMA_VERSIONS = ("cultnet.document_put_raw.v0", "cultnet.document_delete.v0") +TOKEN_FIELD = "sessionToken" +LOOPBACK_HOSTNAMES = ("localhost",) + + +class BindRefused(ValueError): + """A non-loopback bind was requested without the explicit allow.""" + + +def is_loopback_host(host: str) -> bool: + """An empty host means the default 127.0.0.1. A hostname other than localhost is not + resolved and counts as non-loopback, so the allow is needed.""" + host = (host or "127.0.0.1").strip() + if host.lower() in LOOPBACK_HOSTNAMES: + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + +def check_bind(host: str, allow_non_loopback: bool) -> str: + """Return the host to bind, or raise BindRefused for a non-loopback host without the allow.""" + resolved = (host or "127.0.0.1").strip() + if not is_loopback_host(resolved) and not allow_non_loopback: + raise BindRefused( + f"Refusing to bind {resolved!r}: not a loopback address. Tick 'Allow Non-Loopback Bind' " + "in the add-on preferences to serve beyond this machine." + ) + return resolved + + +class SessionGuard: + """The per-session token. Held in memory only; never written to preferences or the cache.""" + + def __init__(self) -> None: + self._token: str | None = secrets.token_urlsafe(32) + + @property + def token(self) -> str: + return self._token or "" + + def revoke(self) -> None: + self._token = None + + def admits(self, message: dict[str, Any]) -> bool: + presented = message.get(TOKEN_FIELD) + if self._token is None or not isinstance(presented, str): + return False + return hmac.compare_digest(presented.encode("utf-8"), self._token.encode("utf-8")) + + +def guarded_server_class(base: type, guard: SessionGuard) -> type: + """A subclass of cultmesh_py's CultMeshLocalServer whose puts and deletes need the guard's token. + + CultMeshLocalServer sends every connection message, TCP and RUDP, through + _handle_connection_message, which is where the check sits. Reads (snapshot, subscribe, + hello) stay open. The refusal is codeless: cultnet.error.v0's code list is closed and + this adds no code to it.""" + + class GuardedServer(base): # type: ignore[misc, valid-type] + def _handle_connection_message(self, message, subscriptions, **kwargs): + if message.get("schemaVersion") in PUT_SCHEMA_VERSIONS and not guard.admits(message): + return [self._error_response( + f"Document writes need the session token in the {TOKEN_FIELD} field.", + message_id=str(message.get("messageId") or ""), + )] + if TOKEN_FIELD in message: + # The store copies every envelope field but schemaVersion into the shard log it + # serves, so the token must not reach it. + message = {key: value for key, value in message.items() if key != TOKEN_FIELD} + return super()._handle_connection_message(message, subscriptions, **kwargs) + + return GuardedServer + + +def purge_command_intents(node: Any, command_document: Any, command_schema: str, debug_mirror_root: str) -> int: + """Delete every queued command intent from the node and every command JSON under + /blender/commands. Returns how many were removed.""" + removed = 0 + if node is not None: + node.database.pull() + for key in list(node.database.snapshot().get(command_schema, {})): + node.database.delete(command_document, key) + removed += 1 + if debug_mirror_root: + for path in (Path(debug_mirror_root) / "blender" / "commands").glob("*.json"): + path.unlink(missing_ok=True) + removed += 1 + return removed diff --git a/surfaces/blender/docs/install.md b/surfaces/blender/docs/install.md index 01a9a85..b76a64e 100644 --- a/surfaces/blender/docs/install.md +++ b/surfaces/blender/docs/install.md @@ -27,6 +27,36 @@ at `0` to bind a free local port, or set a fixed port when another runtime needs a stable endpoint. The panel reports the active `cultnet://host:port` endpoint after `Start Server`. +## Server Safety + +Always on: + +- `Start Server` and `Stop Server` purge queued command intents: every + `brokkr.blender.command_intent.v0` record in the node and every + `blender/commands/*.json` under `Debug Export Root`. Intents queued before a + start never run under the new session. +- After `Start Server` the panel shows the bind address, port and the file + `cultmesh_py` was actually loaded from (its `__file__`, not the preference). +- A `Serve Host` that is not loopback (`0.0.0.0`, a LAN address, a hostname + other than `localhost`) is refused until `Allow Non-Loopback Bind` is ticked in + the add-on preferences by hand. While such a server runs the panel shows a + warning, and unticking the allow stops it. + +Opt-in, default off: `Require Session Token`. This is internal infrastructure +and the machine's owner judged per-write access control not worth the friction +by default. When ticked, `Start Server` generates a random token (`secrets`), +the panel shows it (`Copy Session Token` puts it on the clipboard), and the server rejects every document put and delete whose +message lacks it in a top-level `sessionToken` field. It is checked on every +write, not only at connect. Reads stay open. `Stop Server`, quitting, or +disabling the add-on revokes it; it is held in memory only and never saved to +preferences or the cache. + +Run the tests (no Blender needed; `CULTLIB_PACKAGES` adds the real-server tests): + +```text +CULTLIB_PACKAGES=F:/Projects/CultLib/packages python -m unittest discover -s surfaces/blender/tests -v +``` + `Debug Export Root` may still emit readable JSON probe files under `.brokkr/blender-editor-debug`, but those files are debug/import glue. CultMesh owns the mirror state. diff --git a/surfaces/blender/tests/test_server_guard.py b/surfaces/blender/tests/test_server_guard.py new file mode 100644 index 0000000..296022e --- /dev/null +++ b/surfaces/blender/tests/test_server_guard.py @@ -0,0 +1,339 @@ +"""Server hardening tests. No Blender: the add-on package is registered without running its +__init__ (which imports bpy), so only server_guard and blender_target load. + +Run: python3 -m unittest discover -s surfaces/blender/tests -v +The real-server tests need CultLib's Python packages: set CULTLIB_PACKAGES to the packages +directory, or they are skipped. +""" +from __future__ import annotations + +import os +import sys +import tempfile +import types +import unittest +from pathlib import Path + +ADDON = Path(__file__).resolve().parents[1] / "brokkr_bridge" +package = types.ModuleType("brokkr_bridge") +package.__path__ = [str(ADDON)] +sys.modules.setdefault("brokkr_bridge", package) + +from brokkr_bridge import server_guard as sg # noqa: E402 +from brokkr_bridge import blender_target as bt # noqa: E402 + +PUT = {"schemaVersion": "cultnet.document_put_raw.v0", "messageId": "m1", "document": {}} +DELETE = {"schemaVersion": "cultnet.document_delete.v0", "messageId": "m2"} +SUBSCRIBE = {"schemaVersion": "cultnet.database_subscribe.v0", "messageId": "m3"} + + +class FakeBase: + def __init__(self): + self.handled = [] + + @staticmethod + def _error_response(error, *, message_id="", code=None, details=None): + response = {"schemaVersion": "cultnet.error.v0", "messageId": message_id, "error": error} + if code is not None: + response["code"] = code + return response + + def _handle_connection_message(self, message, subscriptions, *, over_rudp=False): + self.handled.append(message) + return [{"schemaVersion": "handled"}] + + +class SessionGuardTests(unittest.TestCase): + def test_token_is_random_and_unguessable(self): + tokens = {sg.SessionGuard().token for _ in range(50)} + self.assertEqual(len(tokens), 50) + self.assertGreaterEqual(min(len(t) for t in tokens), 32) + + def test_admits_only_the_exact_token(self): + guard = sg.SessionGuard() + self.assertTrue(guard.admits({sg.TOKEN_FIELD: guard.token})) + self.assertFalse(guard.admits({})) + self.assertFalse(guard.admits({sg.TOKEN_FIELD: ""})) + self.assertFalse(guard.admits({sg.TOKEN_FIELD: guard.token + "x"})) + self.assertFalse(guard.admits({sg.TOKEN_FIELD: guard.token[:-1]})) + self.assertFalse(guard.admits({sg.TOKEN_FIELD: 12345})) + self.assertFalse(guard.admits({sg.TOKEN_FIELD: None})) + + def test_revoke_rejects_the_old_token_and_the_empty_one(self): + guard = sg.SessionGuard() + old = guard.token + guard.revoke() + self.assertFalse(guard.admits({sg.TOKEN_FIELD: old})) + self.assertFalse(guard.admits({sg.TOKEN_FIELD: ""})) + self.assertEqual(guard.token, "") + + +class GuardedServerTests(unittest.TestCase): + def setUp(self): + self.guard = sg.SessionGuard() + self.server = sg.guarded_server_class(FakeBase, self.guard)() + + def call(self, message): + return self.server._handle_connection_message(message, {}) + + def test_put_and_delete_without_token_are_refused_and_never_reach_the_store(self): + for message in (PUT, DELETE): + response = self.call(message) + self.assertEqual(response[0]["schemaVersion"], "cultnet.error.v0") + self.assertNotIn("code", response[0]) + self.assertEqual(response[0]["messageId"], message["messageId"]) + self.assertEqual(self.server.handled, []) + + def test_token_is_checked_on_every_put_not_only_the_first(self): + good = {**PUT, sg.TOKEN_FIELD: self.guard.token} + self.assertEqual(self.call(good)[0]["schemaVersion"], "handled") + self.assertEqual(self.call(PUT)[0]["schemaVersion"], "cultnet.error.v0") + self.assertEqual(self.call(good)[0]["schemaVersion"], "handled") + self.assertEqual(self.call({**PUT, sg.TOKEN_FIELD: "wrong"})[0]["schemaVersion"], "cultnet.error.v0") + self.assertEqual(len(self.server.handled), 2) + + def test_token_is_stripped_before_the_store_sees_the_message(self): + for message in (PUT, DELETE): + self.call({**message, sg.TOKEN_FIELD: self.guard.token}) + self.assertEqual(len(self.server.handled), 2) + for handled in self.server.handled: + self.assertNotIn(sg.TOKEN_FIELD, handled) + + def test_reads_need_no_token(self): + self.assertEqual(self.call(SUBSCRIBE)[0]["schemaVersion"], "handled") + + def test_revoked_token_stops_working_on_a_running_server(self): + good = {**PUT, sg.TOKEN_FIELD: self.guard.token} + self.guard.revoke() + self.assertEqual(self.call(good)[0]["schemaVersion"], "cultnet.error.v0") + + +class BindTests(unittest.TestCase): + def test_loopback_hosts_bind_without_the_allow(self): + for host in ("", "127.0.0.1", "127.0.0.2", "::1", "localhost", "LOCALHOST"): + self.assertEqual(sg.check_bind(host, False), host or "127.0.0.1") + + def test_non_loopback_hosts_are_refused_without_the_allow(self): + for host in ("0.0.0.0", "::", "192.168.1.5", "10.0.0.1", "example.com"): + with self.assertRaises(sg.BindRefused): + sg.check_bind(host, False) + + def test_the_allow_lets_a_non_loopback_host_through(self): + self.assertEqual(sg.check_bind("0.0.0.0", True), "0.0.0.0") + self.assertFalse(sg.is_loopback_host("0.0.0.0")) + + +class FakeDatabase: + def __init__(self, schema, keys): + self.schema, self.keys, self.deleted, self.pulls = schema, list(keys), [], 0 + + def pull(self): + self.pulls += 1 + + def snapshot(self): + return {self.schema: {k: {} for k in self.keys}, "other.v0": {"keep": {}}} + + def delete(self, document, key): + self.deleted.append((document, key)) + self.keys.remove(key) + + +class FakeNode: + def __init__(self, keys): + self.database = FakeDatabase(bt.COMMAND_INTENT_SCHEMA, keys) + + +class PurgeTests(unittest.TestCase): + def make_debug_root(self, tmp): + commands = Path(tmp) / "blender" / "commands" + commands.mkdir(parents=True) + (commands / "a.json").write_text("{}") + (commands / "b.json").write_text("{}") + (commands / "note.txt").write_text("keep") + receipts = Path(tmp) / "blender" / "receipts" + receipts.mkdir() + (receipts / "r.json").write_text("{}") + return commands, receipts + + def test_purge_removes_queued_intents_and_debug_command_files_only(self): + with tempfile.TemporaryDirectory() as tmp: + commands, receipts = self.make_debug_root(tmp) + node = FakeNode(["c1", "c2"]) + removed = sg.purge_command_intents(node, "DOC", bt.COMMAND_INTENT_SCHEMA, tmp) + self.assertEqual(removed, 4) + self.assertEqual(node.database.deleted, [("DOC", "c1"), ("DOC", "c2")]) + self.assertEqual(sorted(p.name for p in commands.iterdir()), ["note.txt"]) + self.assertTrue((receipts / "r.json").exists()) + + def test_purge_without_a_debug_root_or_node_is_quiet(self): + self.assertEqual(sg.purge_command_intents(None, "DOC", bt.COMMAND_INTENT_SCHEMA, ""), 0) + + def test_stop_server_purges_and_revokes(self): + with tempfile.TemporaryDirectory() as tmp: + commands, _ = self.make_debug_root(tmp) + target = bt.BrokkrBlenderTarget(types.SimpleNamespace()) + stopped = [] + target._server = types.SimpleNamespace(stop=lambda: stopped.append(True)) + target._guard = sg.SessionGuard() + guard = target._guard + old = guard.token + target._node = FakeNode(["c1"]) + target._documents = {"command_intent": "DOC"} + target._debug_mirror_root = tmp + target.stop_server() + self.assertEqual(stopped, [True]) + self.assertEqual(target._node.database.keys, []) + self.assertEqual(sorted(p.name for p in commands.iterdir()), ["note.txt"]) + self.assertFalse(guard.admits({sg.TOKEN_FIELD: old})) + self.assertEqual(target.session_token, "") + + def test_stop_server_revokes_and_purges_even_when_the_socket_stop_raises(self): + target = bt.BrokkrBlenderTarget(types.SimpleNamespace()) + + def boom(): + raise OSError("close failed") + target._server = types.SimpleNamespace(stop=boom) + target._guard = sg.SessionGuard() + guard = target._guard + old = guard.token + target._node = FakeNode(["c1"]) + target._documents = {"command_intent": "DOC"} + with self.assertRaises(OSError): + target.stop_server() + self.assertFalse(guard.admits({sg.TOKEN_FIELD: old})) + self.assertEqual(target._node.database.keys, []) + self.assertIsNone(target._server) + + def test_enforce_bind_policy_stops_only_a_non_loopback_server_without_the_allow(self): + for host, allow, stays in (("0.0.0.0", False, False), ("0.0.0.0", True, True), ("127.0.0.1", False, True)): + target = bt.BrokkrBlenderTarget(types.SimpleNamespace()) + target._server = types.SimpleNamespace(host=host, stop=lambda: None) + target.enforce_bind_policy(allow) + self.assertEqual(target._server is not None, stays, (host, allow)) + + def test_stop_server_with_no_server_running_purges_nothing(self): + target = bt.BrokkrBlenderTarget(types.SimpleNamespace()) + target._node = FakeNode(["c1"]) + target._documents = {"command_intent": "DOC"} + target.stop_server() + self.assertEqual(target._node.database.keys, ["c1"]) + + +PACKAGES = os.environ.get("CULTLIB_PACKAGES", "") + + +@unittest.skipUnless(PACKAGES and Path(PACKAGES).is_dir(), "set CULTLIB_PACKAGES to CultLib's packages directory") +class RealServerTests(unittest.TestCase): + """The target's start_server against cultmesh_py's real server: token on, token off, bind refusal.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.debug = str(Path(self.tmp.name) / "debug") + commands = Path(self.debug) / "blender" / "commands" + commands.mkdir(parents=True) + (commands / "stale.json").write_text("{}") + self.cache = str(Path(self.tmp.name) / "node.ccmp") + self.target = bt.BrokkrBlenderTarget(types.SimpleNamespace( + path=types.SimpleNamespace(abspath=lambda p: p))) + self.addCleanup(self.target.stop_server) + + def start(self, **kwargs): + return self.target.start_server(self.cache, PACKAGES, "127.0.0.1", 0, 100, 1 << 20, self.debug, **kwargs) + + def test_default_start_has_no_token_and_purges_stale_debug_commands(self): + status = self.start() + self.assertTrue(status["running"]) + self.assertFalse(status["tokenRequired"]) + self.assertEqual(self.target.session_token, "") + self.assertEqual(list((Path(self.debug) / "blender" / "commands").glob("*.json")), []) + self.assertTrue(status["cultmeshModule"].endswith("__init__.py")) + self.assertIn("cultmesh_py", status["cultmeshModule"]) + self.assertTrue(status["loopback"]) + + def test_token_on_refuses_unauthenticated_put_on_the_real_server(self): + status = self.start(require_token=True) + self.assertTrue(status["tokenRequired"]) + server = self.target._server + message = {"schemaVersion": "cultnet.document_put_raw.v0", "messageId": "x", "document": {}} + self.assertEqual(server._handle_connection_message(message, {})[0]["schemaVersion"], "cultnet.error.v0") + message[sg.TOKEN_FIELD] = self.target.session_token + response = server._handle_connection_message(message, {}) + self.assertIn("malformed", response[0].get("code", "")) # past the guard, refused by the store as an empty put + + def queue_intent(self, key="blender/commands/q1"): + node, documents = self.target._open_node(self.cache, PACKAGES) + node.database.put(documents["command_intent"], key, { + "schema": bt.COMMAND_INTENT_SCHEMA, "commandId": "q1", "action": "createObject"}) + return node, documents + + def queued(self, cache_path): + """Intent keys in the .ccmp as a fresh node reads it from disk.""" + cultmesh, cultcache = bt._load_cultmesh(PACKAGES) + fresh = cultmesh.CultMesh.start_node(cache_path, runtime_id="reader") + fresh.database.register_document(cultcache.define_document_type(bt.COMMAND_INTENT_SCHEMA)) + fresh.database.pull() + return list(fresh.database.snapshot().get(bt.COMMAND_INTENT_SCHEMA, {})) + + def test_start_purges_intents_queued_in_the_node_and_they_stay_gone_after_reopening(self): + self.queue_intent() + self.assertEqual(len(self.queued(self.cache)), 1) + self.start() + node, _ = self.target._open_node(self.cache, PACKAGES) + self.assertEqual(list(node.database.snapshot().get(bt.COMMAND_INTENT_SCHEMA, {})), []) + self.assertEqual(self.queued(self.cache), []) + + def test_stop_purges_intents_queued_while_running_and_they_stay_gone_after_reopening(self): + self.start() + self.queue_intent() + self.target.stop_server() + self.assertEqual(self.queued(self.cache), []) + + def test_allow_true_goes_through_start_server_and_unticking_stops_it(self): + status = self.target.start_server(self.cache, PACKAGES, "0.0.0.0", 0, 100, 1 << 20, self.debug, + allow_non_loopback=True) + self.assertTrue(status["running"]) + self.assertFalse(status["loopback"]) + with self.assertRaises(sg.BindRefused): + self.target.start_server(self.cache, PACKAGES, "0.0.0.0", 0, 100, 1 << 20, self.debug, + allow_non_loopback=False) + self.assertFalse(self.target.server_status()["running"]) + + def test_unticking_the_allow_stops_a_running_non_loopback_server(self): + self.target.start_server(self.cache, PACKAGES, "0.0.0.0", 0, 100, 1 << 20, self.debug, + allow_non_loopback=True) + self.target.enforce_bind_policy(False) + self.assertFalse(self.target.server_status()["running"]) + + def test_a_put_with_a_shard_id_and_a_valid_token_leaves_no_token_in_the_shard_log_or_store(self): + import msgpack + status = self.start(require_token=True) + token = self.target.session_token + node, documents = self.target._open_node(self.cache, PACKAGES) + cultmesh, cultcache = bt._load_cultmesh(PACKAGES) + scratch = cultmesh.CultMesh.start_node(str(Path(self.tmp.name) / "scratch.ccmp"), runtime_id="scratch") + scratch.database.register_document(cultcache.define_document_type(bt.COMMAND_INTENT_SCHEMA)) + built = scratch.database.put_raw_message( + cultcache.define_document_type(bt.COMMAND_INTENT_SCHEMA), "blender/commands/s1", + {"schema": bt.COMMAND_INTENT_SCHEMA, "commandId": "s1", "action": "createObject"}, + message_id="m1", shard_id="shard-a", shard_epoch=1) + wire = built.to_wire() + wire[sg.TOKEN_FIELD] = token + server = self.target._server + responses = server._handle_connection_message(wire, {}) + self.assertNotEqual(responses and responses[0].get("schemaVersion"), "cultnet.error.v0", responses) + log = node.database.create_shard_log_response(shard_id="shard-a", shard_epoch=1) + self.assertIn("s1", repr(log)) + self.assertNotIn(token, repr(log)) + self.assertNotIn(sg.TOKEN_FIELD, repr(log)) + self.assertNotIn(token, repr(node.database.snapshot())) + self.assertNotIn(token.encode(), msgpack.packb(log, use_bin_type=True)) + + def test_non_loopback_bind_is_refused_before_anything_starts(self): + with self.assertRaises(sg.BindRefused): + self.target.start_server(self.cache, PACKAGES, "0.0.0.0", 0, 100, 1 << 20, self.debug) + self.assertFalse(self.target.server_status()["running"]) + + +if __name__ == "__main__": + unittest.main()