diff --git a/meshchatx.rsm b/meshchatx.rsm index b691e225..f8632ba3 100644 Binary files a/meshchatx.rsm and b/meshchatx.rsm differ diff --git a/scripts/build_community_interfaces_json.py b/scripts/build_community_interfaces_json.py index 13362174..b4f826df 100644 --- a/scripts/build_community_interfaces_json.py +++ b/scripts/build_community_interfaces_json.py @@ -7,11 +7,11 @@ import argparse import json import sys import urllib.error +import urllib.request from pathlib import Path +from urllib.parse import urlparse from meshchatx.src.backend.community_interfaces_directory import ( - DEFAULT_SUBMITTED_URL, - build_interfaces_from_directory_url, rows_from_payload, transform_directory_rows, ) @@ -19,6 +19,98 @@ from meshchatx.src.backend.community_interfaces_directory import ( ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "meshchatx" / "src" / "backend" / "data" / "community_interfaces.json" +DEFAULT_SUBMITTED_URL = ( + "https://directory.rns.recipes/api/directory/submitted?status=online" +) +DEFAULT_DISCOVERED_URL = ( + "https://directory.rns.recipes/api/directory/discovered?status=online" +) +DEFAULT_DIRECTORY_URLS = (DEFAULT_SUBMITTED_URL, DEFAULT_DISCOVERED_URL) +_ALLOWED_HOST = "directory.rns.recipes" +_MAX_FETCH_BYTES = 1 * 1024 * 1024 +_FETCH_HEADERS = { + "Accept": "application/json", + "User-Agent": "MeshChatX-community-interfaces-build/1.0 (+https://meshchatx.com/)", +} + + +def validate_directory_fetch_url(url: str) -> str: + if not url or not isinstance(url, str): + msg = "URL must be a non-empty string" + raise ValueError(msg) + parsed = urlparse(url.strip()) + if parsed.scheme != "https": + msg = "Community directory URL must use https" + raise ValueError(msg) + netloc = parsed.netloc or "" + if "@" in netloc: + msg = "Community directory URL must not contain credentials" + raise ValueError(msg) + host = (parsed.hostname or "").lower() + if host != _ALLOWED_HOST: + msg = "Community directory URL host is not allowed" + raise ValueError(msg) + return url.strip() + + +def fetch_directory_payload(url: str, *, timeout: float = 60.0) -> object: + resolved = validate_directory_fetch_url(url) + req = urllib.request.Request(resolved, headers=_FETCH_HEADERS, method="GET") + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read(_MAX_FETCH_BYTES + 1) + if len(raw) > _MAX_FETCH_BYTES: + msg = f"Community directory download exceeds {_MAX_FETCH_BYTES} bytes" + raise ValueError(msg) + return json.loads(raw.decode("utf-8")) + + +def _merge_directory_rows(row_lists: list[list]) -> list: + merged: list = [] + seen: set[tuple] = set() + for rows in row_lists: + for row in rows: + if not isinstance(row, dict): + continue + key = ( + str(row.get("name") or "").strip().lower(), + str(row.get("type") or "").strip().lower(), + str(row.get("host") or row.get("address") or "").strip().lower(), + str(row.get("port") or "").strip(), + ) + if key in seen: + continue + seen.add(key) + merged.append(row) + return merged + + +def build_interfaces_from_directory_url( + url: str | None = None, + *, + timeout: float = 60.0, +) -> tuple[list[dict], str]: + if url is not None and str(url).strip(): + resolved = validate_directory_fetch_url(url) + payload = fetch_directory_payload(resolved, timeout=timeout) + rows = rows_from_payload(payload) + return transform_directory_rows(rows), resolved + + row_lists: list[list] = [] + used: list[str] = [] + errors: list[str] = [] + for candidate in DEFAULT_DIRECTORY_URLS: + try: + payload = fetch_directory_payload(candidate, timeout=timeout) + row_lists.append(rows_from_payload(payload)) + used.append(candidate) + except Exception as exc: + errors.append(f"{candidate}: {exc}") + if not used: + msg = "; ".join(errors) if errors else "No directory URLs configured" + raise ValueError(msg) + rows = _merge_directory_rows(row_lists) + return transform_directory_rows(rows), " + ".join(used) + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) diff --git a/tests/backend/http_api_response_registry.py b/tests/backend/http_api_response_registry.py index 3061ecdd..a83b0028 100644 --- a/tests/backend/http_api_response_registry.py +++ b/tests/backend/http_api_response_registry.py @@ -658,9 +658,6 @@ HTTP_JSON_GET_CONTRACT_EXCLUDED: tuple[str, ...] = ( "/api/v1/stickers/export", "/api/v1/sticker-packs/{pack_id}/export", "/api/v1/telephone/contacts/export", - "/api/v1/tools/rnode/download_firmware", - "/api/v1/tools/rnode/latest_release", - "/api/v1/tools/micron-parser-go-release", "/api/v1/filesync/content", "/api/v1/favourites/layout", "/api/v1/map/overlays", diff --git a/tests/backend/test_announce_manager_extended.py b/tests/backend/test_announce_manager_extended.py index 4d5d53ef..0575ce76 100644 --- a/tests/backend/test_announce_manager_extended.py +++ b/tests/backend/test_announce_manager_extended.py @@ -44,6 +44,28 @@ def test_upsert_announce(mock_db): assert data["app_data"] == base64.b64encode(b"app_data").decode("utf-8") +def test_upsert_omits_oversized_app_data(mock_db): + from meshchatx.src.backend.announce_manager import MAX_ANNOUNCE_APP_DATA_BYTES + + manager = AnnounceManager(mock_db) + identity = MagicMock() + identity.hash.hex.return_value = "id_hash" + identity.get_public_key.return_value = b"pub_key" + manager.upsert_announce( + None, + identity, + b"dest_hash", + "aspect", + b"z" * (MAX_ANNOUNCE_APP_DATA_BYTES + 1), + None, + ) + mock_db.announces.upsert_announce.assert_called_once() + data = mock_db.announces.upsert_announce.call_args[0][0] + assert "app_data" not in data + assert data["destination_hash"] == b"dest_hash".hex() + assert data["aspect"] == "aspect" + + def test_upsert_skips_when_store_disabled_for_aspect(mock_db): config = MagicMock() config.announce_store_lxmf_delivery = MagicMock() diff --git a/tests/backend/test_app_security_features.py b/tests/backend/test_app_security_features.py index 16f555ac..c63b3995 100644 --- a/tests/backend/test_app_security_features.py +++ b/tests/backend/test_app_security_features.py @@ -1,7 +1,6 @@ # SPDX-License-Identifier: 0BSD import secrets -from unittest.mock import MagicMock import bcrypt import pytest @@ -154,20 +153,3 @@ async def test_privacy_mode_blocks_map_export(mock_app): headers=headers, ) assert r.status == 403 - - -@pytest.mark.asyncio -@pytest.mark.usefixtures("require_loopback_tcp") -async def test_privacy_mode_blocks_repository_refresh(mock_app): - mock_app.config.privacy_mode_enabled.set(True) - mock_app.repository_server_manager = MagicMock() - aio_app = _make_aio_app(mock_app, use_https=False) - - async with TestClient(TestServer(aio_app)) as client: - headers = await fetch_api_csrf_headers(client) - r = await client.post( - "/api/v1/repository-server/refresh-bundled", - headers=headers, - ) - assert r.status == 403 - mock_app.repository_server_manager.refresh_bundled_wheels.assert_not_called() diff --git a/tests/backend/test_community_interfaces.py b/tests/backend/test_community_interfaces.py index 81300d9c..98d9d68c 100644 --- a/tests/backend/test_community_interfaces.py +++ b/tests/backend/test_community_interfaces.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: 0BSD import json -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock import pytest @@ -22,7 +22,6 @@ async def test_community_interfaces_manager_no_probe(): @pytest.mark.asyncio async def test_rnstatus_integration_simulated(): - # Simulate how rnstatus would see these interfaces if they were added mock_reticulum = MagicMock() mock_reticulum.get_interface_stats.return_value = { "interfaces": [ @@ -61,31 +60,7 @@ async def test_community_interfaces_static_list(): @pytest.mark.asyncio -async def test_community_interfaces_cache_used_when_no_public_override(tmp_path): - cache = tmp_path / "community_interfaces_cache.json" - cache.write_text( - json.dumps( - { - "interfaces": [ - { - "name": "CacheOnly", - "type": "TCPClientInterface", - "target_host": "10.0.0.1", - "target_port": 4242, - }, - ], - }, - ), - encoding="utf-8", - ) - manager = CommunityInterfacesManager(public_override_path=None, cache_path=cache) - ifaces = await manager.get_interfaces() - assert len(ifaces) == 1 - assert ifaces[0]["name"] == "CacheOnly" - - -@pytest.mark.asyncio -async def test_community_interfaces_public_override_beats_cache(tmp_path): +async def test_community_interfaces_public_override_beats_bundled(tmp_path): public = tmp_path / "public.json" public.write_text( json.dumps( @@ -102,50 +77,7 @@ async def test_community_interfaces_public_override_beats_cache(tmp_path): ), encoding="utf-8", ) - cache = tmp_path / "cache.json" - cache.write_text( - json.dumps( - { - "interfaces": [ - { - "name": "FromCache", - "type": "TCPClientInterface", - "target_host": "10.0.0.3", - "target_port": 4242, - }, - ], - }, - ), - encoding="utf-8", - ) - manager = CommunityInterfacesManager( - public_override_path=str(public), - cache_path=str(cache), - ) + manager = CommunityInterfacesManager(public_override_path=str(public)) ifaces = await manager.get_interfaces() assert len(ifaces) == 1 assert ifaces[0]["name"] == "FromPublic" - - -def test_refresh_from_directory_writes_cache(tmp_path): - fake = [ - { - "name": "FromNet", - "type": "TCPClientInterface", - "target_host": "9.9.9.9", - "target_port": 4242, - }, - ] - cache = tmp_path / "community_interfaces_cache.json" - manager = CommunityInterfacesManager(public_override_path=None, cache_path=cache) - with patch( - "meshchatx.src.backend.community_interfaces_directory.build_interfaces_from_directory_url", - return_value=(fake, "https://example.test/list"), - ) as mock_build: - out = manager.refresh_from_directory() - mock_build.assert_called_once() - assert out["count"] == 1 - assert out["source"] == "https://example.test/list" - assert cache.is_file() - manager2 = CommunityInterfacesManager(public_override_path=None, cache_path=cache) - assert manager2.interfaces[0]["name"] == "FromNet" diff --git a/tests/backend/test_community_interfaces_directory.py b/tests/backend/test_community_interfaces_directory.py index dfa2435e..310848d7 100644 --- a/tests/backend/test_community_interfaces_directory.py +++ b/tests/backend/test_community_interfaces_directory.py @@ -1,146 +1,15 @@ # SPDX-License-Identifier: 0BSD -import threading -import urllib.error -import urllib.request -from http.server import BaseHTTPRequestHandler, HTTPServer - import pytest from hypothesis import given, settings from hypothesis import strategies as st from meshchatx.src.backend.community_interfaces_directory import ( - DEFAULT_DIRECTORY_URLS, - DEFAULT_DISCOVERED_URL, - DEFAULT_SUBMITTED_URL, - fetch_directory_payload, rows_from_payload, transform_directory_rows, - validate_directory_fetch_url, ) -def test_default_url_is_submitted_online(): - assert "submitted" in DEFAULT_SUBMITTED_URL - assert "status=online" in DEFAULT_SUBMITTED_URL - assert "discovered" in DEFAULT_DISCOVERED_URL - assert DEFAULT_SUBMITTED_URL in DEFAULT_DIRECTORY_URLS - assert DEFAULT_DISCOVERED_URL in DEFAULT_DIRECTORY_URLS - assert "search=" not in DEFAULT_SUBMITTED_URL - assert "type=" not in DEFAULT_SUBMITTED_URL - - -def test_validate_directory_fetch_url_accepts_default_host(): - u = "https://directory.rns.recipes/api/foo?bar=1" - assert validate_directory_fetch_url(u) == u - - -@pytest.mark.parametrize( - "bad", - [ - "http://directory.rns.recipes/api", - "https://127.0.0.1/", - "https://metadata.internal/", - "ftp://directory.rns.recipes/", - "https://evil.com/https://directory.rns.recipes/", - "https://user:pass@directory.rns.recipes/", - "https://directory.rns.recipes@evil.com/", - "https://not-directory.rns.recipes.example/api", - ], -) -def test_validate_directory_fetch_url_rejects_ssrf(bad): - with pytest.raises(ValueError): - validate_directory_fetch_url(bad) - - -class _Redirect302Handler(BaseHTTPRequestHandler): - def do_GET(self): - self.send_response(302) - self.send_header("Location", "http://127.0.0.1:9/") - self.end_headers() - - def log_message(self, *args): - return - - -class _Json200Handler(BaseHTTPRequestHandler): - def do_GET(self): - body = b'{"data":[]}' - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def log_message(self, *args): - return - - -@pytest.fixture -def redirect_http_port(): - srv = HTTPServer(("127.0.0.1", 0), _Redirect302Handler) - thread = threading.Thread(target=srv.serve_forever, daemon=True) - thread.start() - port = srv.server_address[1] - yield port - srv.shutdown() - - -@pytest.fixture -def json_http_port(): - srv = HTTPServer(("127.0.0.1", 0), _Json200Handler) - thread = threading.Thread(target=srv.serve_forever, daemon=True) - thread.start() - port = srv.server_address[1] - yield port - srv.shutdown() - - -def test_directory_fetch_opener_blocks_http_redirect(redirect_http_port): - from meshchatx.src.backend.community_interfaces_directory import ( - _DIRECTORY_FETCH_OPENER, - ) - - req = urllib.request.Request(f"http://127.0.0.1:{redirect_http_port}/") - with pytest.raises(urllib.error.HTTPError) as ei: - _DIRECTORY_FETCH_OPENER.open(req, timeout=3) - assert ei.value.code == 302 - - -def test_fetch_directory_payload_reads_json_when_no_redirect( - monkeypatch, - json_http_port, -): - import meshchatx.src.backend.community_interfaces_directory as cid - - monkeypatch.setattr( - cid, - "validate_directory_fetch_url", - lambda url: url, - ) - out = fetch_directory_payload( - f"http://127.0.0.1:{json_http_port}/x", - timeout=3, - ) - assert out == {"data": []} - - -def test_fetch_directory_payload_raises_on_redirect(monkeypatch, redirect_http_port): - import meshchatx.src.backend.community_interfaces_directory as cid - - monkeypatch.setattr( - cid, - "validate_directory_fetch_url", - lambda url: url, - ) - with pytest.raises(urllib.error.HTTPError) as ei: - fetch_directory_payload( - f"http://127.0.0.1:{redirect_http_port}/", - timeout=3, - ) - assert ei.value.code == 302 - - def test_rows_from_payload_dict_data(): rows = rows_from_payload({"data": [{"id": 1}]}) assert rows == [{"id": 1}] diff --git a/tests/backend/test_map_geo_validator.py b/tests/backend/test_map_geo_validator.py index e9f070d3..a7b09a56 100644 --- a/tests/backend/test_map_geo_validator.py +++ b/tests/backend/test_map_geo_validator.py @@ -112,6 +112,22 @@ def test_validate_kmz_ok(): assert result.format == "kmz" +def test_validate_kmz_too_large(): + kml = b""" + + 1,2,0 + """ + data = _kmz_with_kml(kml) + with pytest.raises(GeoValidationError) as exc: + validate_geo_bytes( + data, + max_bytes=8, + max_features=100, + max_kmz_uncompressed_bytes=1024 * 1024, + ) + assert exc.value.code == "file_too_large" + + def test_validate_kmz_missing_kml(): buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: diff --git a/tests/backend/test_nomadnet_downloader.py b/tests/backend/test_nomadnet_downloader.py index a9ca6e89..3a5d379f 100644 --- a/tests/backend/test_nomadnet_downloader.py +++ b/tests/backend/test_nomadnet_downloader.py @@ -6,6 +6,7 @@ import pytest import RNS from meshchatx.src.backend.nomadnet_downloader import ( + MAX_NOMAD_PAGE_BYTES, NomadnetDownloader, NomadnetFileDownloader, NomadnetPageDownloader, @@ -139,6 +140,43 @@ def test_page_downloader_empty_response(): on_ok.assert_not_called() +def test_page_downloader_rejects_oversized_body(): + on_ok = MagicMock() + on_fail = MagicMock() + pd = NomadnetPageDownloader( + b"ab" * 8, + "/page.mu", + None, + on_ok, + on_fail, + MagicMock(), + ) + rr = MagicMock() + rr.response = b"x" * (MAX_NOMAD_PAGE_BYTES + 1) + pd.on_download_success(rr) + on_fail.assert_called_once_with("page_too_large") + on_ok.assert_not_called() + + +def test_page_downloader_accepts_body_at_cap(): + on_ok = MagicMock() + on_fail = MagicMock() + pd = NomadnetPageDownloader( + b"ab" * 8, + "/page.mu", + None, + on_ok, + on_fail, + MagicMock(), + ) + rr = MagicMock() + rr.response = b"y" * MAX_NOMAD_PAGE_BYTES + pd.on_download_success(rr) + on_ok.assert_called_once() + on_fail.assert_not_called() + assert on_ok.call_args[0][0] == "y" * MAX_NOMAD_PAGE_BYTES + + def test_file_downloader_list_response_short_list_no_crash(): on_ok = MagicMock() on_fail = MagicMock() diff --git a/tests/backend/test_outbound_http_allowlist.py b/tests/backend/test_outbound_http_allowlist.py new file mode 100644 index 00000000..765e3871 --- /dev/null +++ b/tests/backend/test_outbound_http_allowlist.py @@ -0,0 +1,74 @@ +# SPDX-License-Identifier: 0BSD + +"""New backend clearnet fetches must be listed here and gated by privacy mode. + +Do not add a file to KNOWN_CLEARNET_FETCH_FILES until it calls +ensure_outbound_http_allowed, http_url_guard, or _require_outbound_http +(or is the RNS HTTPInterface, which is mesh transport, not app clearnet). +""" + +from pathlib import Path + +BACKEND_ROOT = Path("meshchatx/src/backend") + +FETCH_MARKERS = ( + "aiohttp.ClientSession", + "httpx.Client(", + "httpx.AsyncClient(", + "urllib.request.urlopen", + "urllib.request.Request(", +) + +KNOWN_CLEARNET_FETCH_FILES = frozenset( + { + "data/interfaces/HTTPInterface.py", + "map_manager.py", + "repository_server_manager.py", + "translator_handler.py", + }, +) + +MESH_TRANSPORT_FETCH_FILES = frozenset( + { + "data/interfaces/HTTPInterface.py", + }, +) + +PRIVACY_GATE_MARKERS = ( + "ensure_outbound_http_allowed", + "http_url_guard", + "_require_outbound_http", + "OutboundHttpBlockedError", +) + + +def _rel(path: Path) -> str: + return str(path.relative_to(BACKEND_ROOT)).replace("\\", "/") + + +def test_new_backend_clearnet_fetches_are_allowlisted(): + found: set[str] = set() + for path in BACKEND_ROOT.rglob("*.py"): + text = path.read_text(encoding="utf-8") + if any(marker in text for marker in FETCH_MARKERS): + found.add(_rel(path)) + assert found == set(KNOWN_CLEARNET_FETCH_FILES), ( + "New backend httpx/urllib/aiohttp client usage must call " + "ensure_outbound_http_allowed or http_url_guard, then be added to " + "KNOWN_CLEARNET_FETCH_FILES. Extra: " + f"{sorted(found - KNOWN_CLEARNET_FETCH_FILES)}. Missing from disk: " + f"{sorted(KNOWN_CLEARNET_FETCH_FILES - found)}" + ) + + +def test_allowlisted_app_fetches_mention_a_privacy_gate(): + for rel in sorted(KNOWN_CLEARNET_FETCH_FILES - MESH_TRANSPORT_FETCH_FILES): + text = (BACKEND_ROOT / rel).read_text(encoding="utf-8") + if any(marker in text for marker in PRIVACY_GATE_MARKERS): + continue + # Route-gated helpers: the HTTP route must still call the gate. + assert rel in { + "map_manager.py", + "repository_server_manager.py", + "translator_handler.py", + }, f"{rel} opens a clearnet socket without a privacy-mode or URL guard" diff --git a/tests/backend/test_repository_server_manager.py b/tests/backend/test_repository_server_manager.py index 03fead60..be0a1f26 100644 --- a/tests/backend/test_repository_server_manager.py +++ b/tests/backend/test_repository_server_manager.py @@ -2,7 +2,6 @@ import time import urllib.request -from pathlib import Path from unittest.mock import patch import pytest @@ -153,18 +152,6 @@ def test_save_rejects_invalid_upload_filenames(tmp_path, name): assert not ok -@patch( - "meshchatx.src.backend.repository_server_manager.download_bundled_wheels_to_directory", -) -def test_refresh_invokes_bundled_downloader(mock_dl, tmp_path): - mock_dl.return_value = {"ok": True, "downloaded": ["rns"], "failed": {}} - mgr = RepositoryServerManager(str(tmp_path)) - out = mgr.refresh_bundled_wheels() - assert out["ok"] is True - mock_dl.assert_called_once() - assert mock_dl.call_args.kwargs.get("on_package") is not None - - @patch( "meshchatx.src.backend.repository_server_manager.stage_local_meshchatx_wheel_into_bundled_dir", return_value=None, @@ -209,58 +196,6 @@ def test_download_bundled_wheels_records_pypi_failures( assert mock_pypi.call_count == n -@patch( - "meshchatx.src.backend.repository_server_manager.stage_local_meshchatx_wheel_into_bundled_dir", - return_value=None, -) -@patch("meshchatx.src.backend.repository_server_manager._download_wheel_via_pypi_index") -def test_refresh_bundled_wheels_fails_when_pypi_unavailable( - mock_pypi, - _mock_stage, - tmp_path, - monkeypatch, -): - monkeypatch.setenv("MESHCHAT_REPOSITORY_EXTRA_PIP", "") - mock_pypi.return_value = (False, "offline") - mgr = RepositoryServerManager(str(tmp_path)) - out = mgr.refresh_bundled_wheels() - assert out["ok"] is False - assert not out["downloaded"] - - -@patch( - "meshchatx.src.backend.repository_server_manager.stage_local_meshchatx_wheel_into_bundled_dir", - return_value=None, -) -@patch("meshchatx.src.backend.repository_server_manager._download_wheel_via_pypi_index") -def test_refresh_preserves_existing_wheels_when_pypi_fails( - mock_pypi, - _mock_stage, - tmp_path, - monkeypatch, -): - monkeypatch.setenv("MESHCHAT_REPOSITORY_EXTRA_PIP", "") - mock_pypi.return_value = (False, "offline") - mgr = RepositoryServerManager(str(tmp_path)) - keep = Path(mgr.bundled_dir) / "keep-me.whl" - keep.write_bytes(b"wheel") - out = mgr.refresh_bundled_wheels() - assert out["ok"] is False - assert keep.exists() - assert keep.read_bytes() == b"wheel" - - -def test_refresh_rejects_concurrent_calls(tmp_path): - mgr = RepositoryServerManager(str(tmp_path)) - assert mgr._refresh_lock.acquire(blocking=False) - try: - out = mgr.refresh_bundled_wheels() - assert out["ok"] is False - assert out.get("error") == "refresh_already_running" - finally: - mgr._refresh_lock.release() - - def test_http_start_stop_and_status(tmp_path): mgr = RepositoryServerManager(str(tmp_path)) assert mgr.status()["http"]["running"] is False diff --git a/tests/backend/test_rnode_download_firmware.py b/tests/backend/test_rnode_download_firmware.py deleted file mode 100644 index 218dfc24..00000000 --- a/tests/backend/test_rnode_download_firmware.py +++ /dev/null @@ -1,418 +0,0 @@ -# SPDX-License-Identifier: 0BSD - -"""HTTP integration tests for the RNode firmware proxy endpoint.""" - -from __future__ import annotations - -from contextlib import asynccontextmanager -from unittest.mock import MagicMock, patch - -import pytest -from aiohttp import web -from aiohttp.test_utils import TestClient, TestServer - -pytestmark = pytest.mark.usefixtures("require_loopback_tcp") - - -def _build_aio_app(app): - routes = web.RouteTableDef() - auth_mw, mime_mw, sec_mw, csrf_mw, ip_mw, demo_mw = app._define_routes(routes) - aio_app = web.Application( - middlewares=[auth_mw, mime_mw, sec_mw, csrf_mw, ip_mw, demo_mw] - ) - aio_app.add_routes(routes) - return aio_app - - -@pytest.fixture -def web_app(mock_app): - mock_app.current_context.running = True - mock_app.config.auth_enabled.set(False) - return mock_app - - -class _FakeResponse: - def __init__(self, status: int, body: bytes, url: str = "https://github.com/x"): - self.status = status - self._body = body - self.url = url - - async def read(self): - return self._body - - -class _FakeSession: - def __init__(self, status: int, body: bytes, final_url: str | None = None): - self._status = status - self._body = body - self._final_url = final_url - self.requested_urls: list[str] = [] - - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc, tb): - return False - - def get(self, url, allow_redirects=True, headers=None): - self.requested_urls.append(url) - status = self._status - body = self._body - final_url = self._final_url or url - - @asynccontextmanager - async def _cm(): - yield _FakeResponse(status, body, url=final_url) - - return _cm() - - -@pytest.mark.asyncio -async def test_download_firmware_requires_url(web_app): - aio_app = _build_aio_app(web_app) - async with TestClient(TestServer(aio_app)) as client: - r = await client.get("/api/v1/tools/rnode/download_firmware") - assert r.status == 400 - body = await r.json() - assert "URL" in body["error"] - - -@pytest.mark.asyncio -async def test_download_firmware_rejects_disallowed_redirect_target(web_app): - aio_app = _build_aio_app(web_app) - fake_session = _FakeSession( - 200, - b"PK\x03\x04ssrf", - final_url="http://127.0.0.1:9337/secret", - ) - - with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={ - "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip", - }, - ) - assert r.status == 403 - body = await r.json() - assert "redirect" in body["error"].lower() - - -@pytest.mark.asyncio -async def test_download_firmware_allows_codeload_redirect(web_app): - aio_app = _build_aio_app(web_app) - fake_zip = b"PK\x03\x04ok" - final = "https://codeload.github.com/owner/repo/zip/refs/tags/v1" - fake_session = _FakeSession(200, fake_zip, final_url=final) - - with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={ - "url": "https://github.com/owner/repo/archive/refs/tags/v1.zip", - }, - ) - assert r.status == 200 - assert await r.read() == fake_zip - - -@pytest.mark.asyncio -async def test_download_firmware_returns_zip_for_allowed_url(web_app): - aio_app = _build_aio_app(web_app) - fake_zip = b"PK\x03\x04fake-zip-bytes" - fake_session = _FakeSession(200, fake_zip) - - with patch( - "aiohttp.ClientSession", - MagicMock(return_value=fake_session), - ): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={ - "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip", - }, - ) - assert r.status == 200 - assert r.headers.get("Content-Type", "").startswith("application/zip") - assert r.headers.get("Content-Disposition", "").endswith('"firmware.zip"') - data = await r.read() - assert data == fake_zip - assert fake_session.requested_urls == [ - "https://github.com/owner/repo/releases/download/v1/firmware.zip", - ] - - -@pytest.mark.asyncio -async def test_download_firmware_strips_crlf_from_content_disposition(web_app): - aio_app = _build_aio_app(web_app) - fake_zip = b"PK\x03\x04hdr" - fake_session = _FakeSession(200, fake_zip) - hostile_name = 'evil.zip"\r\nX-Injected: yes' - url = "https://github.com/owner/repo/releases/download/v1/" + hostile_name - - with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={"url": url}, - ) - assert r.status == 200 - disp = r.headers.get("Content-Disposition", "") - prefix = 'attachment; filename="' - assert disp.startswith(prefix) - assert disp.endswith('"') - inner = disp[len(prefix) : -1] - assert "\r" not in disp - assert "\n" not in disp - assert '"' not in inner - assert await r.read() == fake_zip - - -@pytest.mark.asyncio -async def test_download_firmware_propagates_upstream_error_status(web_app): - aio_app = _build_aio_app(web_app) - fake_session = _FakeSession(404, b"") - - with patch( - "aiohttp.ClientSession", - MagicMock(return_value=fake_session), - ): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={ - "url": "https://github.com/markqvist/RNode_Firmware/releases/download/v1/firmware.zip", - }, - ) - assert r.status == 404 - body = await r.json() - assert "Failed to download" in body["error"] - - -@pytest.mark.asyncio -async def test_download_firmware_returns_500_on_exception(web_app): - aio_app = _build_aio_app(web_app) - - with patch( - "aiohttp.ClientSession", - MagicMock(side_effect=RuntimeError("network down")), - ): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={ - "url": "https://github.com/owner/repo/releases/download/v1/firmware.zip", - }, - ) - assert r.status == 500 - body = await r.json() - assert "network down" in body["error"] - - -class _FakeJsonResponse: - def __init__(self, status: int, payload): - self.status = status - self._payload = payload - - async def json(self, content_type=None): - return self._payload - - async def read(self): - import json - - return json.dumps(self._payload).encode("utf-8") - - -class _FakeJsonSession: - def __init__(self, status: int, payload): - self._status = status - self._payload = payload - self.requested_urls: list[str] = [] - self.last_headers = None - - async def __aenter__(self): - return self - - async def __aexit__(self, exc_type, exc, tb): - return False - - def get(self, url, allow_redirects=True, headers=None): - self.requested_urls.append(url) - self.last_headers = headers - status = self._status - payload = self._payload - - @asynccontextmanager - async def _cm(): - yield _FakeJsonResponse(status, payload) - - return _cm() - - -@pytest.mark.asyncio -async def test_download_firmware_accepts_objects_githubusercontent_url(web_app): - aio_app = _build_aio_app(web_app) - fake_zip = b"PK\x03\x04x" - fake_session = _FakeSession(200, fake_zip) - asset_url = ( - "https://objects.githubusercontent.com/github-production-release-asset/1/2/3" - "?response-content-disposition=attachment%3B%20filename%3Dfw.zip" - ) - - with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={"url": asset_url}, - ) - assert r.status == 200 - assert fake_session.requested_urls == [asset_url] - - -@pytest.mark.asyncio -async def test_download_firmware_accepts_release_assets_githubusercontent_url(web_app): - aio_app = _build_aio_app(web_app) - fake_zip = b"PK\x03\x04y" - fake_session = _FakeSession(200, fake_zip) - asset_url = "https://release-assets.githubusercontent.com/github-production-release-asset/9/8/7/fw.zip" - - with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={"url": asset_url}, - ) - assert r.status == 200 - data = await r.read() - assert data == fake_zip - - -@pytest.mark.asyncio -async def test_download_firmware_accepts_configured_gitea_base_url(web_app): - aio_app = _build_aio_app(web_app) - web_app.config.gitea_base_url.set("https://gitea.custom.example") - fake_zip = b"PK\x03\x04z" - fake_session = _FakeSession(200, fake_zip) - asset_url = ( - "https://gitea.custom.example/someorg/somerepo/releases/download/v1/x.zip" - ) - - with patch("aiohttp.ClientSession", MagicMock(return_value=fake_session)): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/download_firmware", - params={"url": asset_url}, - ) - assert r.status == 200 - assert fake_session.requested_urls == [asset_url] - - -@pytest.mark.asyncio -async def test_latest_release_returns_proxied_payload(web_app): - aio_app = _build_aio_app(web_app) - payload = { - "tag_name": "v1.83", - "assets": [ - { - "name": "rnode_firmware_heltec32v3.zip", - "browser_download_url": "https://x/rnode.zip", - }, - ], - } - fake_session = _FakeJsonSession(200, payload) - - with patch( - "aiohttp.ClientSession", - MagicMock(return_value=fake_session), - ): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get("/api/v1/tools/rnode/latest_release") - assert r.status == 200 - body = await r.json() - assert body == payload - assert fake_session.requested_urls[0] == ( - "https://api.github.com/repos/markqvist/RNode_Firmware/releases/latest" - ) - assert fake_session.last_headers is not None - assert ( - fake_session.last_headers.get("Accept") == "application/vnd.github+json" - ) - assert fake_session.last_headers.get("X-GitHub-Api-Version") == "2022-11-28" - assert "MeshChatX-RNodeFlasher" in fake_session.last_headers.get( - "User-Agent", - "", - ) - - -@pytest.mark.asyncio -async def test_latest_release_uses_repo_query_param(web_app): - aio_app = _build_aio_app(web_app) - fake_session = _FakeJsonSession(200, {"tag_name": "v0"}) - - with patch( - "aiohttp.ClientSession", - MagicMock(return_value=fake_session), - ): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get( - "/api/v1/tools/rnode/latest_release", - params={"repo": "Some/Other_Repo"}, - ) - assert r.status == 200 - assert fake_session.requested_urls[0] == ( - "https://api.github.com/repos/Some/Other_Repo/releases/latest" - ) - - -@pytest.mark.asyncio -async def test_latest_release_rejects_invalid_repo(web_app): - aio_app = _build_aio_app(web_app) - async with TestClient(TestServer(aio_app)) as client: - for repo in ( - "no-slash", - "../etc/passwd", - "evil repo/x", - "bad?repo/x", - "too/many/slashes", - "@bad/name", - "/leading/slash", - "trailing/", - ): - r = await client.get( - "/api/v1/tools/rnode/latest_release", - params={"repo": repo}, - ) - assert r.status == 400, f"expected 400 for repo={repo!r}" - - -@pytest.mark.asyncio -async def test_latest_release_propagates_upstream_status(web_app): - aio_app = _build_aio_app(web_app) - fake_session = _FakeJsonSession(404, {}) - - with patch( - "aiohttp.ClientSession", - MagicMock(return_value=fake_session), - ): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get("/api/v1/tools/rnode/latest_release") - assert r.status == 404 - body = await r.json() - assert "Failed to fetch release" in body["error"] - - -@pytest.mark.asyncio -async def test_latest_release_returns_500_on_exception(web_app): - aio_app = _build_aio_app(web_app) - with patch( - "aiohttp.ClientSession", - MagicMock(side_effect=RuntimeError("dns down")), - ): - async with TestClient(TestServer(aio_app)) as client: - r = await client.get("/api/v1/tools/rnode/latest_release") - assert r.status == 500 - body = await r.json() - assert "dns down" in body["error"] diff --git a/tests/backend/test_rnode_download_url_oracle.py b/tests/backend/test_rnode_download_url_oracle.py deleted file mode 100644 index d35890b4..00000000 --- a/tests/backend/test_rnode_download_url_oracle.py +++ /dev/null @@ -1,32 +0,0 @@ -# SPDX-License-Identifier: 0BSD - -"""Oracle: RNode firmware download URL allowlist rejects lookalike hosts.""" - -from __future__ import annotations - - -def _url_allowed(url: str, allowed_prefixes: list[str]) -> bool: - return any(url.startswith(a) for a in allowed_prefixes) - - -def test_rnode_download_prefix_oracle_rejects_lookalike_hosts(): - allowed = [ - "https://github.com/", - "https://codeload.github.com/", - "https://objects.githubusercontent.com/", - "https://release-assets.githubusercontent.com/", - ] - # Accept - assert _url_allowed( - "https://github.com/markqvist/RNode_Firmware/releases/download/v1/x.zip", - allowed, - ) - assert _url_allowed( - "https://objects.githubusercontent.com/github-production-release-asset-2e65be/1/x", - allowed, - ) - # Reject lookalikes / SSRF bait - assert not _url_allowed("https://github.com.evil.example/markqvist/x.zip", allowed) - assert not _url_allowed("https://evil.example/https://github.com/x.zip", allowed) - assert not _url_allowed("http://127.0.0.1/firmware.zip", allowed) - assert not _url_allowed("https://github.com.attacker/x", allowed) diff --git a/tests/backend/test_rrc_security.py b/tests/backend/test_rrc_security.py index be8422f3..9eaad9ea 100644 --- a/tests/backend/test_rrc_security.py +++ b/tests/backend/test_rrc_security.py @@ -233,6 +233,32 @@ def test_server_rejects_oversized_message_body(): assert "large" in out[0][1][proto.K_BODY].lower() +def test_server_rejects_oversized_message_body_at_default_cap(): + server = make_server() + assert server.max_msg_body_bytes == proto.DEFAULT_MAX_MSG_BYTES + link = FakeLink(FakeIdentity(b"\x01" * 16)) + sess = add_session(server, link, link._identity.hash, nick="alice") + route( + server, + link, + sess, + proto.make_envelope(proto.T_JOIN, src=sess.peer, room="lobby"), + ) + out = route( + server, + link, + sess, + proto.make_envelope( + proto.T_MSG, + src=sess.peer, + room="lobby", + body="x" * (proto.DEFAULT_MAX_MSG_BYTES + 1), + ), + ) + assert out[0][1][proto.K_T] == proto.T_ERROR + assert "large" in out[0][1][proto.K_BODY].lower() + + def test_server_rejects_message_to_unknown_room(): server = make_server() link = FakeLink(FakeIdentity(b"\x02" * 16))