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