1 files changed, 230 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_tmdb_config_policy.py b/packages/meshbay-node/tests/test_tmdb_config_policy.py
new file mode 100644
index 0000000..29ef54c
--- /dev/null
+++ b/packages/meshbay-node/tests/test_tmdb_config_policy.py
@@ -0,0 +1,230 @@
+"""
+The operator decides whether the node calls TMDB at all, and whether it uses
+a custom API token — docs/mediacenter.md §5.5. Same shape as
+test_apps_enabled_policy.py/test_scan_settings_policy.py: a signed operator
+instruction, node-wide (group_id="") rather than per-group, stored via
+roster.py's group_settings table.
+
+Specific to this one: the subject signed/audited must never contain the
+token itself (it would end up in the audit log in plaintext) — only whether
+one was supplied travels there. The token itself only ever travels in
+`payload`, which is node-side context never re-sent or re-verified from the
+wire (see _issue_admin_challenge's docstring).
+"""
+
+from pathlib import Path
+
+import pytest
+
+from meshbay_common.adminop import OP_TMDB_CONFIG
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": one_root(shared_root),
+        "index": index,
+        "sk_node": index.sk_node,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = None
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _fake_challenge(issued: list):
+    return lambda op, subject, payload=None, group_id=None: issued.append(
+        (op, subject, payload, group_id))
+
+
+# ── Refused before a challenge is even issued ───────────────────────────────
+
+async def test_missing_enabled_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_non_bool_enabled_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({"enabled": "yes"})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_non_string_token_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({"enabled": True, "token": 12345})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    session = _session(tmp_path, "member-1", operator="the-operator")
+    session._has_admin_authority = lambda: False
+
+    session._do_tmdb_config({"enabled": False})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Who may change it, and what gets signed ─────────────────────────────────
+
+async def test_changing_it_needs_a_signature(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({"enabled": True})
+
+    assert len(issued) == 1
+    op, subject, payload, group_id = issued[0]
+    assert op == OP_TMDB_CONFIG
+    assert group_id == "", "node-wide, like group_attach/group_detach — not tied to self._group_id"
+
+
+async def test_the_token_itself_never_appears_in_the_signed_subject(tmp_path):
+    """The subject is what gets audited (self._audit(pending['subject'])) —
+    a secret must never end up there in plaintext."""
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    secret = "sk-super-secret-tmdb-token"
+    session._do_tmdb_config({"enabled": True, "token": secret})
+
+    _, subject, payload, _ = issued[0]
+    assert secret not in subject
+    assert payload["token"] == secret, "the real value still has to reach the exec step somehow"
+
+
+async def test_subject_reflects_enabled_and_whether_a_token_was_supplied(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({"enabled": False, "token": "x"})
+
+    _, subject, _, _ = issued[0]
+    assert subject == "enabled=False,custom_token=yes,language=default"
+
+
+async def test_subject_says_no_custom_token_when_none_given(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({"enabled": True})
+
+    _, subject, _, _ = issued[0]
+    assert subject == "enabled=True,custom_token=no,language=default"
+
+
+async def test_subject_reflects_a_configured_language(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({"enabled": True, "language": "fr-FR"})
+
+    _, subject, payload, _ = issued[0]
+    assert subject == "enabled=True,custom_token=no,language=fr-FR"
+    assert payload["language"] == "fr-FR"
+
+
+async def test_non_string_language_is_refused(tmp_path):
+    session = _session(tmp_path, "op", operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = _fake_challenge(issued)
+
+    session._do_tmdb_config({"enabled": True, "language": 42})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        enabled, token, language = await roster.tmdb_config()
+        assert (enabled, token, language) == (True, None, None), (
+            "absent must mean on, with the shipped default token, TMDB's own default language")
+        await roster.set_tmdb_config(True, "my-custom-token", "fr-FR", set_by="op")
+        enabled, token, language = await roster.tmdb_config()
+        assert (enabled, token, language) == (True, "my-custom-token", "fr-FR")
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.tmdb_config() == (True, "my-custom-token", "fr-FR")
+    finally:
+        await reopened.close()
+
+
+async def test_clearing_the_token_reverts_to_the_default(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_tmdb_config(True, "a-token", set_by="op")
+        assert (await roster.tmdb_config())[1] == "a-token"
+
+        await roster.set_tmdb_config(True, "", set_by="op")
+        enabled, token, language = await roster.tmdb_config()
+        assert token is None, "an explicit empty string clears the custom token"
+    finally:
+        await roster.close()
+
+
+async def test_omitting_the_token_leaves_it_unchanged(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_tmdb_config(True, "a-token", set_by="op")
+        await roster.set_tmdb_config(False, None, set_by="op")
+        enabled, token, language = await roster.tmdb_config()
+        assert (enabled, token) == (False, "a-token")
+    finally:
+        await roster.close()