diff --git a/pyproject.toml b/pyproject.toml index 44b4693..13e4ba2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "uipath-dev" -version = "0.0.82" +version = "0.0.83" description = "UiPath Developer Console" readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.11" diff --git a/src/uipath/dev/server/app.py b/src/uipath/dev/server/app.py index 6c7b0d6..085f47b 100644 --- a/src/uipath/dev/server/app.py +++ b/src/uipath/dev/server/app.py @@ -12,6 +12,10 @@ from fastapi.responses import HTMLResponse from uipath.dev.server import UiPathDeveloperServer +from uipath.dev.server.security import ( + LoopbackGuardMiddleware, + loopback_origin_regex, +) logger = logging.getLogger(__name__) @@ -103,12 +107,15 @@ def create_app(server: UiPathDeveloperServer) -> FastAPI: app.add_middleware( CORSMiddleware, - allow_origins=["*"], + allow_origin_regex=loopback_origin_regex(), allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) + # Added last so it wraps CORS and runs first on every request and WebSocket. + app.add_middleware(LoopbackGuardMiddleware) + # Store server reference on app state for route access app.state.server = server diff --git a/src/uipath/dev/server/routes/files.py b/src/uipath/dev/server/routes/files.py index d69b667..96a3a40 100644 --- a/src/uipath/dev/server/routes/files.py +++ b/src/uipath/dev/server/routes/files.py @@ -84,6 +84,20 @@ def _resolve_safe(path: str) -> Path: return resolved +def _reject_if_secret(target: Path) -> None: + """Block reads/writes of credential files inside the project root.""" + rel_parts = target.relative_to(ROOT).parts if target.is_relative_to(ROOT) else () + if ".uipath" in rel_parts: + raise HTTPException( + status_code=403, detail="Access to this file is not allowed" + ) + name = target.name + if name == ".env" or name.startswith(".env."): + raise HTTPException( + status_code=403, detail="Access to this file is not allowed" + ) + + def _detect_language(filepath: Path) -> str | None: name = filepath.name.lower() if name == "dockerfile": @@ -139,6 +153,7 @@ async def list_directory(path: str = "") -> list[dict[str, Any]]: async def read_file(path: str) -> dict[str, Any]: """Read file content. Returns content, binary flag, size, and language.""" target = _resolve_safe(path) + _reject_if_secret(target) if not target.is_file(): raise HTTPException(status_code=404, detail="File not found") @@ -189,6 +204,7 @@ class SaveFileBody(BaseModel): async def save_file(path: str, body: SaveFileBody) -> dict[str, str]: """Save file content. Creates parent dirs if needed.""" target = _resolve_safe(path) + _reject_if_secret(target) target.parent.mkdir(parents=True, exist_ok=True) # Write bytes directly to avoid Python text-mode converting \n → \r\n on # Windows, which would add extra carriage returns on every save since diff --git a/src/uipath/dev/server/security.py b/src/uipath/dev/server/security.py new file mode 100644 index 0000000..9597f1f --- /dev/null +++ b/src/uipath/dev/server/security.py @@ -0,0 +1,80 @@ +"""Loopback protection for the unauthenticated developer server.""" + +from __future__ import annotations + +from typing import Any + +from starlette.types import ASGIApp, Receive, Scope, Send + +_LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1"} + + +def _hostname(header_value: str) -> str: + """Extract the hostname from a Host/Origin header, dropping scheme and port.""" + value = header_value.strip() + if "://" in value: + value = value.split("://", 1)[1] + if value.startswith("["): + return value[1 : value.index("]")] if "]" in value else value[1:] + return value.rsplit(":", 1)[0] if ":" in value else value + + +class LoopbackGuardMiddleware: + """Reject requests whose Host or Origin is not loopback.""" + + def __init__(self, app: ASGIApp) -> None: + """Wrap the downstream ASGI app.""" + self.app = app + + def _allowed_hosts(self, scope: Scope) -> set[str]: + allowed = set(_LOOPBACK_HOSTS) + server = getattr(scope.get("app"), "state", None) + configured = getattr(getattr(server, "server", None), "host", None) + if configured: + allowed.add(configured) + return allowed + + def _is_allowed(self, scope: Scope) -> bool: + headers = {k.lower(): v for k, v in scope.get("headers", [])} + allowed = self._allowed_hosts(scope) + + host = headers.get(b"host") + if host is not None and _hostname(host.decode("latin-1")) not in allowed: + return False + + origin = headers.get(b"origin") + if origin is not None and _hostname(origin.decode("latin-1")) not in allowed: + return False + + return True + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + """Guard HTTP and WebSocket scopes, pass everything else through.""" + if scope["type"] not in ("http", "websocket") or self._is_allowed(scope): + await self.app(scope, receive, send) + return + + if scope["type"] == "websocket": + await send({"type": "websocket.close", "code": 1008}) + return + + await self._forbidden(send) + + async def _forbidden(self, send: Send) -> None: + body = b"Forbidden: request rejected by loopback guard" + await send( + { + "type": "http.response.start", + "status": 403, + "headers": [ + (b"content-type", b"text/plain; charset=utf-8"), + (b"content-length", str(len(body)).encode("latin-1")), + ], + } + ) + await send({"type": "http.response.body", "body": body}) + + +def loopback_origin_regex(*args: Any) -> str: + """Regex matching loopback origins for CORS ``allow_origin_regex``.""" + return r"^https?://(localhost|127\.0\.0\.1|\[::1\])(:\d+)?$" diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 0000000..4cbc835 --- /dev/null +++ b/tests/test_security.py @@ -0,0 +1,116 @@ +"""Security tests for the developer server loopback protections. + +Covers VULN-45558: the unauthenticated file/run APIs must reject cross-site +(``Origin``) and DNS-rebinding (``Host``) requests, and must never serve the +UiPath Cloud access token stored in ``.env`` / ``.uipath``. +""" + +import pytest +from fastapi import FastAPI, WebSocket +from starlette.testclient import TestClient + +from uipath.dev.server.security import ( + LoopbackGuardMiddleware, + _hostname, + loopback_origin_regex, +) + + +def _app() -> FastAPI: + app = FastAPI() + app.add_middleware(LoopbackGuardMiddleware) + + @app.get("/api/ping") + async def ping() -> dict[str, str]: + return {"status": "ok"} + + @app.websocket("/ws") + async def ws(websocket: WebSocket) -> None: + await websocket.accept() + await websocket.send_text("ok") + await websocket.close() + + return app + + +def _client() -> TestClient: + return TestClient(_app(), base_url="http://localhost") + + +@pytest.mark.parametrize("host", ["localhost", "127.0.0.1", "localhost:8080"]) +def test_loopback_host_allowed(host: str) -> None: + resp = _client().get("/api/ping", headers={"Host": host}) + assert resp.status_code == 200 + + +@pytest.mark.parametrize("host", ["evil.com", "attacker.example:8080", "169.254.1.1"]) +def test_non_loopback_host_rejected(host: str) -> None: + resp = _client().get("/api/ping", headers={"Host": host}) + assert resp.status_code == 403 + + +def test_cross_site_origin_rejected() -> None: + resp = _client().get("/api/ping", headers={"Origin": "https://evil.example"}) + assert resp.status_code == 403 + + +def test_same_origin_allowed() -> None: + resp = _client().get("/api/ping", headers={"Origin": "http://localhost:8080"}) + assert resp.status_code == 200 + + +def test_no_origin_allowed() -> None: + assert _client().get("/api/ping").status_code == 200 + + +def test_websocket_rejects_cross_site_origin() -> None: + from starlette.websockets import WebSocketDisconnect + + with pytest.raises(WebSocketDisconnect): + with _client().websocket_connect( + "/ws", headers={"Host": "localhost", "Origin": "https://evil.example"} + ) as ws: + ws.receive_text() + + +def test_websocket_allows_loopback() -> None: + with _client().websocket_connect("/ws", headers={"Host": "localhost"}) as ws: + assert ws.receive_text() == "ok" + + +@pytest.mark.parametrize( + "value,expected", + [ + ("http://evil.example:8080", "evil.example"), + ("localhost:8080", "localhost"), + ("[::1]:8080", "::1"), + ("127.0.0.1", "127.0.0.1"), + ], +) +def test_hostname_parsing(value: str, expected: str) -> None: + assert _hostname(value) == expected + + +def test_cors_regex_matches_loopback_only() -> None: + import re + + pattern = re.compile(loopback_origin_regex()) + assert pattern.match("http://localhost:8080") + assert pattern.match("http://127.0.0.1:8080") + assert not pattern.match("https://evil.example") + + +def test_secret_files_are_blocked(monkeypatch) -> None: + from fastapi import HTTPException + + from uipath.dev.server.routes import files + + for name in (".env", ".env.local"): + with pytest.raises(HTTPException) as exc: + files._reject_if_secret(files.ROOT / name) + assert exc.value.status_code == 403 + + with pytest.raises(HTTPException): + files._reject_if_secret(files.ROOT / ".uipath" / ".auth.json") + + files._reject_if_secret(files.ROOT / "main.py") diff --git a/uv.lock b/uv.lock index dbe6d4a..c386558 100644 --- a/uv.lock +++ b/uv.lock @@ -2592,7 +2592,7 @@ wheels = [ [[package]] name = "uipath-dev" -version = "0.0.82" +version = "0.0.83" source = { editable = "." } dependencies = [ { name = "aiosqlite" },