Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -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"
Expand Down
9 changes: 8 additions & 1 deletion src/uipath/dev/server/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)

Expand Down Expand Up @@ -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

Expand Down
16 changes: 16 additions & 0 deletions src/uipath/dev/server/routes/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand Down Expand Up @@ -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")

Expand Down Expand Up @@ -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
Expand Down
80 changes: 80 additions & 0 deletions src/uipath/dev/server/security.py
Original file line number Diff line number Diff line change
@@ -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+)?$"
116 changes: 116 additions & 0 deletions tests/test_security.py
Original file line number Diff line number Diff line change
@@ -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")
2 changes: 1 addition & 1 deletion uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading