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,

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]:

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
+    <debug_mirror_root>/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

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.

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()