diff --git a/frontends/desktop_bridge.py b/frontends/desktop_bridge.py index a42388d8f..de9dcb741 100644 --- a/frontends/desktop_bridge.py +++ b/frontends/desktop_bridge.py @@ -1399,26 +1399,78 @@ async def ws_handler(request): # Transport layer: HTTP command/data API # --------------------------------------------------------------------------- -def cors_headers(): +_LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"} + + +def _origin_parts(origin: str): + if origin != origin.strip(): return None + from urllib.parse import urlsplit + try: + parsed = urlsplit(origin) + parsed.port + except ValueError: + return None + if (parsed.scheme.lower() not in ("http", "https") or not parsed.hostname or + parsed.username is not None or parsed.password is not None or + parsed.path or parsed.query or parsed.fragment): + return None + return parsed + + +def _request_host_parts(host: str): + from urllib.parse import urlsplit + try: + parsed = urlsplit("//" + host) + parsed.port + except ValueError: + return None + if parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment: + return None + return parsed if parsed.hostname else None + + +def _origin_matches_host(request, origin: str) -> bool: + if not origin: + return True + parsed_origin = _origin_parts(origin) + parsed_host = _request_host_parts(request.host) + if parsed_origin is None or parsed_host is None: + return False + origin_host = parsed_origin.hostname.lower() + request_host = parsed_host.hostname.lower() + bind_host = os.environ.get("BRIDGE_HOST", "127.0.0.1").strip().lower() + if bind_host in _LOOPBACK_HOSTS: + return origin_host in _LOOPBACK_HOSTS and request_host in _LOOPBACK_HOSTS + return parsed_origin.netloc.lower() == request.host.lower() + + +def cors_headers(origin: str = ""): + if not origin: + return {} return { - "Access-Control-Allow-Origin": "*", - "Access-Control-Allow-Methods": "GET,POST,PUT,DELETE,OPTIONS", + "Access-Control-Allow-Origin": origin, + "Access-Control-Allow-Methods": "GET,POST,PUT,PATCH,DELETE,OPTIONS", "Access-Control-Allow-Headers": "Content-Type", + "Vary": "Origin", } @web.middleware async def cors_middleware(request, handler): + origin = request.headers.get("Origin", "") + if origin and not _origin_matches_host(request, origin): + return web.json_response({"ok": False, "error": "cross-origin request rejected"}, status=403) + headers = cors_headers(origin) if request.method == "OPTIONS": - return web.Response(status=204, headers=cors_headers()) + return web.Response(status=204, headers=headers) resp = await handler(request) - for k, v in cors_headers().items(): + for k, v in headers.items(): resp.headers[k] = v return resp def json_ok(data: dict, status: int = 200): - return web.json_response(data, status=status, headers=cors_headers(), dumps=lambda x: json.dumps(x, ensure_ascii=False, default=str)) + return web.json_response(data, status=status, dumps=lambda x: json.dumps(x, ensure_ascii=False, default=str)) async def read_json(request) -> dict: diff --git a/tests/test_desktop_bridge_origin.py b/tests/test_desktop_bridge_origin.py new file mode 100644 index 000000000..8ea0be9cb --- /dev/null +++ b/tests/test_desktop_bridge_origin.py @@ -0,0 +1,116 @@ +import importlib.util +import os +import sys +import tempfile +import unittest +from pathlib import Path + +from aiohttp import web +from aiohttp.test_utils import TestClient, TestServer + + +ROOT = Path(__file__).resolve().parents[1] +FRONTENDS = ROOT / "frontends" +if str(FRONTENDS) not in sys.path: + sys.path.insert(0, str(FRONTENDS)) + +# desktop_bridge has import-time persistence/upload setup. Point it at an isolated +# valid GA root so test discovery cannot touch a developer's configured GA_ROOT. +_TEST_GA_ROOT = tempfile.TemporaryDirectory() +_TEST_GA_PATH = Path(_TEST_GA_ROOT.name) +(_TEST_GA_PATH / "agentmain.py").touch() +_old_ga_root = os.environ.get("GA_ROOT") +_old_argv = sys.argv[:] +os.environ["GA_ROOT"] = str(_TEST_GA_PATH) +sys.argv = [sys.argv[0]] +try: + spec = importlib.util.spec_from_file_location("desktop_bridge", FRONTENDS / "desktop_bridge.py") + assert spec and spec.loader + bridge = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = bridge + spec.loader.exec_module(bridge) +finally: + sys.argv = _old_argv + if _old_ga_root is None: + os.environ.pop("GA_ROOT", None) + else: + os.environ["GA_ROOT"] = _old_ga_root + + +class DesktopBridgeOriginTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + app = web.Application(middlewares=[bridge.cors_middleware]) + + async def probe(request): + return bridge.json_ok({"ok": True}) + + app.router.add_post("/probe", probe) + self.client = TestClient(TestServer(app)) + await self.client.start_server() + + async def asyncTearDown(self): + await self.client.close() + + async def test_import_uses_isolated_ga_root(self): + self.assertEqual(Path(bridge.DEFAULT_GA_ROOT), _TEST_GA_PATH) + self.assertEqual(bridge._WEB_UPLOAD_DIR, _TEST_GA_PATH / "temp" / "desktop_uploads") + + async def test_cross_origin_preflight_is_rejected(self): + response = await self.client.options( + "/probe", + headers={ + "Origin": "https://evil.example", + "Access-Control-Request-Method": "POST", + "Access-Control-Request-Headers": "content-type", + }, + ) + self.assertEqual(response.status, 403) + + async def test_cross_origin_post_is_rejected(self): + response = await self.client.post( + "/probe", headers={"Origin": "https://evil.example"}, json={} + ) + self.assertEqual(response.status, 403) + + async def test_dns_rebinding_origin_is_rejected_on_loopback(self): + response = await self.client.post( + "/probe", + headers={"Host": "evil.example:14168", "Origin": "https://evil.example:14168"}, + json={}, + ) + self.assertEqual(response.status, 403) + + async def test_malformed_serialized_origins_are_rejected(self): + origin = str(self.client.make_url("/")).rstrip("/") + origins = ( + f"{origin}/path", + f"{origin}?x=1", + origin.replace("http://", "http://user@", 1), + "http://127.0.0.1:notaport", + f"{origin}#fragment", + ) + for candidate in origins: + with self.subTest(origin=candidate): + response = await self.client.post( + "/probe", headers={"Origin": candidate}, json={} + ) + self.assertEqual(response.status, 403) + + async def test_null_origin_is_rejected(self): + response = await self.client.post("/probe", headers={"Origin": "null"}, json={}) + self.assertEqual(response.status, 403) + + async def test_same_origin_request_is_allowed(self): + origin = str(self.client.make_url("/")).rstrip("/") + response = await self.client.post("/probe", headers={"Origin": origin}, json={}) + self.assertEqual(response.status, 200) + self.assertEqual(response.headers.get("Access-Control-Allow-Origin"), origin) + + async def test_non_browser_request_without_origin_is_allowed(self): + response = await self.client.post("/probe", json={}) + self.assertEqual(response.status, 200) + self.assertNotIn("Access-Control-Allow-Origin", response.headers) + + +if __name__ == "__main__": + unittest.main()