1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
"""
`tmdb_rematch` (§10.1/V13) — an operator dropping one file's cached TMDB
match so it re-resolves with the current matcher. Signed like
`tmdb_override` (media_cache is shared node-wide); unlike `clear_file_tmdb`
it forgets a manual override marker too, since the operator is explicitly
asking for a fresh resolution.
"""
import hashlib
import pytest
from conftest import one_root
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.adminop import OP_TMDB_REMATCH
from meshbay_common.protocol import MNP, IndexEntry
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.media_cache import MediaCache
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
pytestmark = pytest.mark.asyncio
def _session(tmp_path, user_id, *, operator=None):
shared = tmp_path / "shared"
shared.mkdir(exist_ok=True)
index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
s = WebRTCPeerSession.__new__(WebRTCPeerSession)
s._ctx = {"roots": one_root(shared), "index": index, "sk_node": index.sk_node,
"node_user_id": operator}
s._group_id = None
s._user_id = user_id
s._pk_user = ""
s.sent = []
s._send = s.sent.append
s._audit = lambda *a, **k: None
return s
def _entry(name):
digest = hashlib.sha256(name.encode()).hexdigest()
return IndexEntry(id=digest, name=name, path="movies", size=1, type="video",
added_at=0, display_title="Some Film")
async def _true():
return True
async def test_missing_file_id_is_refused(tmp_path):
s = _session(tmp_path, "op", operator="op")
s._has_admin_authority = lambda: True
issued = []
s._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
s._do_tmdb_rematch({})
assert not issued
assert [m for m in s.sent if m.get("type") == "error"]
async def test_unknown_file_id_is_refused(tmp_path):
s = _session(tmp_path, "op", operator="op")
s._has_admin_authority = lambda: True
issued = []
s._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
s._do_tmdb_rematch({"file_id": "nope"})
assert not issued
assert [m for m in s.sent if m.get("type") == "error"]
async def test_no_authorized_key_is_refused(tmp_path):
s = _session(tmp_path, "member", operator="the-operator")
e = _entry("some.film.2001.mkv")
s._ctx["index"].add_entry(e)
s._has_admin_authority = lambda: False
s._do_tmdb_rematch({"file_id": e.id})
assert [m for m in s.sent if m.get("type") == "error"]
async def test_a_valid_request_is_signed(tmp_path):
s = _session(tmp_path, "op", operator="op")
e = _entry("some.film.2001.mkv")
s._ctx["index"].add_entry(e)
s._has_admin_authority = lambda: True
issued = []
s._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
s._do_tmdb_rematch({"file_id": e.id})
assert issued == [(OP_TMDB_REMATCH, f"file_id={e.id}")]
async def test_exec_drops_the_match_and_the_override_marker(tmp_path):
s = _session(tmp_path, "op", operator="op")
e = _entry("some.film.2001.mkv")
s._ctx["index"].add_entry(e)
media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
await media_cache.open()
try:
s._ctx["media_cache"] = media_cache
await media_cache.set_file_tmdb(e.id, "wrong-id", "movie")
await media_cache.mark_tmdb_override(e.id)
s._verify_admin_sig = lambda transcript, sig: _true()
peer = type("Peer", (), {"sent": []})()
peer._send = peer.sent.append
s._peer_registry = lambda: {"peer-1": peer}
await s._admin_exec_tmdb_rematch(
{"subject": f"file_id={e.id}"}, b"transcript", b"sig")
assert await media_cache.get_file_tmdb(e.id) is None
# the override marker is gone too, so a later group-wide rematch
# would treat a fresh match as ordinary
await media_cache.set_file_tmdb(e.id, "fresh", "movie")
assert await media_cache.clear_tmdb_matches([e.id]) == 1
assert [m for m in peer.sent if m.get("type") == MNP.TMDB_REMATCH_ACK]
finally:
await media_cache.close()
async def test_exec_refuses_a_bad_signature(tmp_path):
s = _session(tmp_path, "op", operator="op")
e = _entry("some.film.2001.mkv")
s._ctx["index"].add_entry(e)
media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
await media_cache.open()
try:
s._ctx["media_cache"] = media_cache
await media_cache.set_file_tmdb(e.id, "keep-me", "movie")
async def _false():
return False
s._verify_admin_sig = lambda transcript, sig: _false()
await s._admin_exec_tmdb_rematch(
{"subject": f"file_id={e.id}"}, b"transcript", b"badsig")
assert await media_cache.get_file_tmdb(e.id) == ("keep-me", "movie")
assert [m for m in s.sent if m.get("type") == "error"]
finally:
await media_cache.close()
|