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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
"""
Integration test: ChunkServer ↔ ChunkClient over TLS.
Starts a real TLS server on localhost, connects a client,
fetches index and a chunk, verifies signature+hash+decryption.
"""
import asyncio
import base64
import os
import time
import jwt
import pytest
from pathlib import Path
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
from meshbay_common.crypto import generate_gek, pk_to_b64
from meshbay_node.indexer import DirectoryIndexer, GroupIndex
from meshbay_node.transport.server import ChunkServer
from meshbay_node.transport.client import ChunkClient
@pytest.fixture
def sk_node():
return Ed25519PrivateKey.generate()
@pytest.fixture
def sk_hub():
return Ed25519PrivateKey.generate()
@pytest.fixture
def gek():
return generate_gek()
@pytest.fixture
def shared_dir(tmp_path):
d = tmp_path / "shared"
d.mkdir()
(d / "test.mp4").write_bytes(os.urandom(2 * 1024 * 1024)) # 2 MB
(d / "small.txt").write_bytes(b"hello meshbay " * 100)
return d
def make_jwt(sk_hub, pk_node_b64, user_id="user-001", ttl=3600):
sk_pem = sk_hub.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
)
now = int(time.time())
return jwt.encode({
"iss": "test-hub", "sub": user_id,
"pk_user": pk_node_b64, "hub_id": "test-hub",
"jti": "test-jti",
"iat": now, "exp": now + ttl,
}, sk_pem, algorithm="EdDSA")
@pytest.mark.asyncio
async def test_chunk_server_client_roundtrip(
sk_node, sk_hub, gek, shared_dir, tmp_path):
"""Full integration: server serves a chunk, client verifies and decrypts."""
# Build index
indexer = DirectoryIndexer(
root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
await indexer.initial_scan()
assert indexer.index.count == 2
# Hub PK for JWT verification
hub_pk_pem = sk_hub.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo)
# TLS cert in tmp dir
cert_path = tmp_path / "node.crt"
key_path = tmp_path / "node.key"
server = ChunkServer(
sk_node=sk_node,
hub_pk_pem=hub_pk_pem,
gek=gek,
shared_root=shared_dir,
index=indexer.index,
host="127.0.0.1",
port=0, # OS picks a free port
cert_path=cert_path,
key_path=key_path,
)
await server.start()
port = server._server.sockets[0].getsockname()[1]
token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()))
# Find the large test file in the index
entry = next(e for e in indexer.index.entries if e.name == "test.mp4")
async with ChunkClient(
host="127.0.0.1",
port=port,
jwt_token=token,
gek=gek,
pk_node_b64=pk_to_b64(sk_node.public_key()),
) as client:
# Fetch first chunk
chunk0 = await client.fetch_chunk(entry.id, chunk_index=0)
assert len(chunk0) == 1024 * 1024 # first 1MB of 2MB file
# Fetch second chunk
chunk1 = await client.fetch_chunk(entry.id, chunk_index=1)
assert len(chunk1) == 1024 * 1024 # second 1MB
# Reassembled file matches original
original = (shared_dir / "test.mp4").read_bytes()
assert chunk0 + chunk1 == original
await server.stop()
@pytest.mark.asyncio
async def test_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path):
hub_pk_pem = sk_hub.public_key().public_bytes(
serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
indexer = DirectoryIndexer(root=shared_dir, group_id="g",
sk_node=sk_node, gek=gek)
await indexer.initial_scan()
cert_path = tmp_path / "node.crt"
key_path = tmp_path / "node.key"
server = ChunkServer(
sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
shared_root=shared_dir, index=indexer.index,
host="127.0.0.1", port=0,
cert_path=cert_path, key_path=key_path,
)
await server.start()
port = server._server.sockets[0].getsockname()[1]
# Use a different hub key to sign the token
sk_other_hub = Ed25519PrivateKey.generate()
bad_token = make_jwt(sk_other_hub, pk_to_b64(sk_node.public_key()))
with pytest.raises(Exception):
async with ChunkClient(
host="127.0.0.1", port=port,
jwt_token=bad_token, gek=gek,
pk_node_b64=pk_to_b64(sk_node.public_key()),
) as client:
pass
await server.stop()
@pytest.mark.asyncio
async def test_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
hub_pk_pem = sk_hub.public_key().public_bytes(
serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
indexer = DirectoryIndexer(root=shared_dir, group_id="g",
sk_node=sk_node, gek=gek)
await indexer.initial_scan()
cert_path = tmp_path / "node.crt"
key_path = tmp_path / "node.key"
server = ChunkServer(
sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
shared_root=shared_dir, index=indexer.index,
host="127.0.0.1", port=0,
cert_path=cert_path, key_path=key_path,
)
await server.start()
port = server._server.sockets[0].getsockname()[1]
token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()))
async with ChunkClient(
host="127.0.0.1", port=port, jwt_token=token,
gek=gek, pk_node_b64=pk_to_b64(sk_node.public_key()),
) as client:
wire = await client.fetch_index()
recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
assert recovered.count == 2
await server.stop()
|