aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/tests/test_http_server.py
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-node/tests/test_http_server.py')
-rw-r--r--packages/meshbay-node/tests/test_http_server.py227
1 files changed, 0 insertions, 227 deletions
diff --git a/packages/meshbay-node/tests/test_http_server.py b/packages/meshbay-node/tests/test_http_server.py
deleted file mode 100644
index d4ccc32..0000000
--- a/packages/meshbay-node/tests/test_http_server.py
+++ /dev/null
@@ -1,227 +0,0 @@
-"""Tests for the node HTTP file API."""
-
-import asyncio
-import base64
-import json
-import os
-import time
-import pytest
-import jwt
-import httpx
-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
-from meshbay_node.transport.http_server import create_http_app
-
-
-@pytest.fixture
-def sk_node():
- return Ed25519PrivateKey.generate()
-
-@pytest.fixture
-def sk_hub():
- return Ed25519PrivateKey.generate()
-
-@pytest.fixture
-def hub_pk_pem(sk_hub):
- return sk_hub.public_key().public_bytes(
- serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
-
-@pytest.fixture
-def gek():
- return generate_gek()
-
-@pytest.fixture
-def shared_dir(tmp_path):
- d = tmp_path / "shared"
- d.mkdir()
- (d / "video.mp4").write_bytes(os.urandom(3 * 1024 * 1024)) # 3MB
- (d / "doc.pdf").write_bytes(os.urandom(512 * 1024))
- (d / "song.mp3").write_bytes(os.urandom(256 * 1024))
- return d
-
-def make_token(sk_hub, pk_node_b64, 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-001",
- "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_node_info(sk_node, sk_hub, hub_pk_pem, gek, shared_dir):
- indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
- await indexer.initial_scan()
-
- app = create_http_app(
- sk_node=sk_node, hub_pk_pem=hub_pk_pem,
- shared_root=shared_dir, index=indexer.index,
- group_id="test-group", group_name="Test Group",
- )
- async with httpx.AsyncClient(
- transport=httpx.ASGITransport(app=app), base_url="http://test"
- ) as c:
- r = await c.get("/")
- assert r.status_code == 200
- data = r.json()
- assert data["group_id"] == "test-group"
- assert data["file_count"] == 3
- assert "pk_node" in data
-
-
-@pytest.mark.asyncio
-async def test_public_index(sk_node, sk_hub, hub_pk_pem, shared_dir):
- """Public group: index accessible without auth."""
- indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
- await indexer.initial_scan()
-
- app = create_http_app(
- sk_node=sk_node, hub_pk_pem=hub_pk_pem,
- shared_root=shared_dir, index=indexer.index,
- group_id="pub-group", group_name="Public Group",
- gek=None,
- )
- async with httpx.AsyncClient(
- transport=httpx.ASGITransport(app=app), base_url="http://test"
- ) as c:
- r = await c.get("/index")
- assert r.status_code == 200
- data = r.json()
- assert len(data["entries"]) == 3
- names = {e["name"] for e in data["entries"]}
- assert "video.mp4" in names
- assert "doc.pdf" in names
-
-
-@pytest.mark.asyncio
-async def test_file_download(sk_node, sk_hub, hub_pk_pem, shared_dir):
- """Full file download via HTTP."""
- indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
- await indexer.initial_scan()
-
- app = create_http_app(
- sk_node=sk_node, hub_pk_pem=hub_pk_pem,
- shared_root=shared_dir, index=indexer.index,
- group_id="g", group_name="G",
- )
- entry = next(e for e in indexer.index.entries if e.name == "doc.pdf")
- original = (shared_dir / "doc.pdf").read_bytes()
-
- async with httpx.AsyncClient(
- transport=httpx.ASGITransport(app=app), base_url="http://test"
- ) as c:
- r = await c.get(f"/file/{entry.id}")
- assert r.status_code == 200
- assert r.content == original
-
-
-@pytest.mark.asyncio
-async def test_chunk_public_group(sk_node, sk_hub, hub_pk_pem, shared_dir):
- """Public group chunk: plaintext, signed, auth required."""
- indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
- await indexer.initial_scan()
-
- app = create_http_app(
- sk_node=sk_node, hub_pk_pem=hub_pk_pem,
- shared_root=shared_dir, index=indexer.index,
- group_id="g", group_name="G", gek=None,
- )
- entry = next(e for e in indexer.index.entries if e.name == "video.mp4")
- token = make_token(sk_hub, pk_to_b64(sk_node.public_key()))
-
- async with httpx.AsyncClient(
- transport=httpx.ASGITransport(app=app), base_url="http://test"
- ) as c:
- r = await c.get(f"/file/{entry.id}/0",
- headers={"Authorization": f"Bearer {token}"})
- assert r.status_code == 200
- chunk = r.json()
- assert chunk["encrypted"] is False
- assert chunk["chunk_index"] == 0
- assert "data_b64" in chunk
-
- # Verify the chunk data matches original
- original = (shared_dir / "video.mp4").read_bytes()
- data = base64.b64decode(chunk["data_b64"])
- assert data == original[:len(data)]
-
-
-@pytest.mark.asyncio
-async def test_chunk_private_group(sk_node, sk_hub, hub_pk_pem, gek, shared_dir):
- """Private group chunk: encrypted with GEK."""
- from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk
- import blake3
-
- indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
- await indexer.initial_scan()
-
- app = create_http_app(
- sk_node=sk_node, hub_pk_pem=hub_pk_pem,
- shared_root=shared_dir, index=indexer.index,
- group_id="g", group_name="G", gek=gek,
- )
- entry = next(e for e in indexer.index.entries if e.name == "doc.pdf")
- token = make_token(sk_hub, pk_to_b64(sk_node.public_key()))
-
- async with httpx.AsyncClient(
- transport=httpx.ASGITransport(app=app), base_url="http://test"
- ) as c:
- r = await c.get(f"/file/{entry.id}/0",
- headers={"Authorization": f"Bearer {token}"})
- assert r.status_code == 200
- chunk = r.json()
- assert chunk["encrypted"] is True
-
- # Decrypt and verify
- file_hash = base64.b64decode(chunk["file_hash_b64"])
- nonce = base64.b64decode(chunk["nonce_b64"])
- ct = base64.b64decode(chunk["ct_b64"])
- ckey = derive_chunk_key(gek, file_hash, 0)
- plaintext = decrypt_chunk(ckey, nonce, ct)
- original = (shared_dir / "doc.pdf").read_bytes()
- assert plaintext == original[:len(plaintext)]
-
-
-@pytest.mark.asyncio
-async def test_chunk_requires_auth(sk_node, hub_pk_pem, shared_dir):
- """Chunk endpoint rejects unauthenticated requests."""
- indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
- await indexer.initial_scan()
- app = create_http_app(
- sk_node=sk_node, hub_pk_pem=hub_pk_pem,
- shared_root=shared_dir, index=indexer.index,
- group_id="g", group_name="G",
- )
- entry = indexer.index.entries[0]
-
- async with httpx.AsyncClient(
- transport=httpx.ASGITransport(app=app), base_url="http://test"
- ) as c:
- r = await c.get(f"/file/{entry.id}/0") # no token
- assert r.status_code == 401
-
-
-@pytest.mark.asyncio
-async def test_unknown_file_404(sk_node, hub_pk_pem, sk_hub, shared_dir):
- indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
- await indexer.initial_scan()
- app = create_http_app(
- sk_node=sk_node, hub_pk_pem=hub_pk_pem,
- shared_root=shared_dir, index=indexer.index,
- group_id="g", group_name="G",
- )
- token = make_token(sk_hub, pk_to_b64(sk_node.public_key()))
-
- async with httpx.AsyncClient(
- transport=httpx.ASGITransport(app=app), base_url="http://test"
- ) as c:
- r = await c.get("/file/nonexistent-hash/0",
- headers={"Authorization": f"Bearer {token}"})
- assert r.status_code == 404
generated by cgit v1.2.3 (git 2.51.0) at 2026-09-25 18:01:24 +0000