Skip to content
Merged
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
101 changes: 101 additions & 0 deletions openswarm-edge/app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,15 @@
"OPENSWARM_CLOUD_INTERNAL_URL", "http://openswarm-cloud.internal:8080"
).rstrip("/")
EDGE_AUTH_TOKEN = os.environ.get("EDGE_AUTH_TOKEN", "")
EDGE_PUBLIC_HOST = os.environ.get(
"EDGE_PUBLIC_HOST", f"{os.environ.get('FLY_APP_NAME', 'openswarm-edge')}.fly.dev"
).lower()

_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,31}$")
_DOWNLOAD_FILENAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,180}$")
_llm_limiter = RateLimiter(limit=30, window_seconds=60)
_compute_limiter = RateLimiter(limit=60, window_seconds=60)
_download_limiter = RateLimiter(limit=30, window_seconds=60)

app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)

Expand Down Expand Up @@ -147,6 +152,102 @@ async def relay():
)


@app.get("/download/{platform}/{arch}")
async def edge_download(platform: str, arch: str, request: Request) -> Response:
# This is an edge-app route, not an app-bundle route. Keeping the exact host
# check prevents it from leaking onto every {slug}.openswarm.host domain.
request_host = request.headers.get("host", "").split(":", 1)[0].lower()
if request_host != EDGE_PUBLIC_HOST:
return HTMLResponse(not_found_page(), status_code=404)
if not _download_limiter.allow(client_ip(request)):
return JSONResponse({"error": "Too many download requests, slow down."}, status_code=429)

ref = (request.query_params.get("ref") or "").strip()
authorize_body = {
"platform": platform,
"arch": arch,
**({"ref": ref} if ref else {}),
}
try:
async with httpx.AsyncClient(timeout=10.0) as client:
authorization = await client.post(
f"{CLOUD_INTERNAL_URL}/api/install/authorize-download",
headers={"x-edge-auth": EDGE_AUTH_TOKEN},
json=authorize_body,
)
except httpx.HTTPError:
return JSONResponse({"error": "Download authorization is unavailable."}, status_code=502)

if authorization.status_code != 200:
# Public validation errors are safe and useful to preserve. Internal
# authentication/configuration failures are exposed only as a 502.
if authorization.status_code in (400, 404, 429):
try:
detail = authorization.json()
message = detail.get("message") or detail.get("error")
except (ValueError, AttributeError):
message = None
return JSONResponse(
{"error": message or "Download request was rejected."},
status_code=authorization.status_code,
)
return JSONResponse({"error": "Download authorization failed."}, status_code=502)

try:
download = authorization.json()
except ValueError:
return JSONResponse({"error": "Download authorization returned invalid data."}, status_code=502)
asset_url = download.get("assetUrl") if isinstance(download, dict) else None
filename = download.get("filename") if isinstance(download, dict) else None
if (
not isinstance(asset_url, str)
or not asset_url.startswith(("https://", "http://"))
or not isinstance(filename, str)
or not _DOWNLOAD_FILENAME_RE.fullmatch(filename)
):
return JSONResponse({"error": "Download authorization returned invalid data."}, status_code=502)

client = httpx.AsyncClient(timeout=httpx.Timeout(60.0, read=None), follow_redirects=True)
upstream_req = client.build_request(
"GET", asset_url, headers={"User-Agent": "openswarm-edge-download"}
)
try:
upstream = await client.send(upstream_req, stream=True)
except httpx.HTTPError:
await client.aclose()
return JSONResponse({"error": "Release artifact is unavailable."}, status_code=502)
if not upstream.is_success:
status = upstream.status_code
await upstream.aclose()
await client.aclose()
return JSONResponse(
{"error": f"Release artifact is unavailable ({status})."}, status_code=502
)

async def relay():
try:
async for chunk in upstream.aiter_raw():
yield chunk
finally:
await upstream.aclose()
await client.aclose()

headers = {
"Content-Disposition": f'attachment; filename="{filename}"',
"Cache-Control": "private, no-store" if ref else "public, max-age=300",
}
content_length = upstream.headers.get("content-length")
if content_length:
headers["Content-Length"] = content_length

return StreamingResponse(
relay(),
status_code=upstream.status_code,
media_type=upstream.headers.get("content-type", "application/octet-stream"),
headers=headers,
)


@app.get("/{path:path}")
async def serve_static(path: str, request: Request) -> Response:
slug = slug_from_host(request.headers.get("host", ""))
Expand Down
8 changes: 4 additions & 4 deletions openswarm-edge/app/ratelimit.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Tiny in-memory per-key fixed-window rate limiter. Guards /__compute and /__llm
so one visitor or scraper can't burn a creator's budget or our CPU. Best-effort
and single-process: the cloud's budget ledger is the hard backstop, this just
keeps the obvious abuse out cheaply."""
"""Tiny in-memory per-key fixed-window rate limiter. Guards /__compute, /__llm,
and installer downloads so one visitor or scraper can't burn a creator's budget,
our CPU, or artifact egress. Best-effort and single-process: the cloud's budget
ledger is the hard backstop for metered calls."""
from __future__ import annotations

import time
Expand Down
1 change: 1 addition & 0 deletions openswarm-edge/fly.toml
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ kill_timeout = '30s'
[env]
PORT = '8080'
APPS_BASE_DOMAIN = 'openswarm.host'
EDGE_PUBLIC_HOST = 'openswarm-edge.fly.dev'
# Private 6PN address of the cloud app (same org). Not public.
OPENSWARM_CLOUD_INTERNAL_URL = 'http://openswarm-cloud.internal:8080'
TIGRIS_ENDPOINT = 'https://fly.storage.tigris.dev'
Expand Down
109 changes: 106 additions & 3 deletions openswarm-edge/tests/test_edge.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,22 @@
"""Unit tests for the edge's pure logic: Host->slug parsing, path-safe file
resolution, the rate limiter, and the vendored sandbox. The Tigris fetch + cloud
proxy need live services and are exercised in the staging E2E, not here.
"""Unit tests for edge routing, path-safe bundle resolution, rate limiting,
installer streaming with mocked upstreams, and the vendored sandbox. Tigris and
the LLM proxy still require live services and are exercised in staging E2E.

Run with: .venv/bin/python -m pytest tests/test_edge.py
"""
import asyncio
import io
import json
import os
import sys
import tarfile

import httpx
from fastapi.testclient import TestClient

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

from app import main as edge_main
from app.main import slug_from_host
from app.bundles import unpack, resolve_file
from app.inject import inject_runtime
Expand Down Expand Up @@ -67,6 +72,104 @@ def test_rate_limiter():
assert rl.allow("ip2") is True # a different key is independent


def _mock_download_clients(monkeypatch, *, reject_unknown=False):
calls = []

class InstallerStream(httpx.AsyncByteStream):
async def __aiter__(self):
yield b"installer-bytes"

async def handle(request: httpx.Request) -> httpx.Response:
calls.append(request)
if request.url.path == "/api/install/authorize-download":
payload = json.loads(request.content)
assert request.headers["x-edge-auth"] == edge_main.EDGE_AUTH_TOKEN
if reject_unknown and payload["platform"] == "plan9":
return httpx.Response(404, json={"message": "unknown download target"})
return httpx.Response(
200,
json={
"assetUrl": "https://github.test/OpenSwarm-arm64.dmg",
"filename": "OpenSwarm-arm64-affiliate_hash_123.dmg",
},
)
if request.url.host == "github.test":
return httpx.Response(
200,
stream=InstallerStream(),
headers={
"Content-Type": "application/x-apple-diskimage",
"Content-Length": "15",
},
)
raise AssertionError(f"unexpected upstream request: {request.url}")

transport = httpx.MockTransport(handle)
real_async_client = httpx.AsyncClient

def mocked_async_client(*args, **kwargs):
kwargs["transport"] = transport
return real_async_client(*args, **kwargs)

monkeypatch.setattr(edge_main.httpx, "AsyncClient", mocked_async_client)
return calls


def test_download_streams_bytes_with_cloud_filename(monkeypatch):
calls = _mock_download_clients(monkeypatch)
client = TestClient(edge_main.app)
response = client.get(
"/download/mac/arm64?ref=alice",
headers={"host": edge_main.EDGE_PUBLIC_HOST, "fly-client-ip": "198.51.100.11"},
)

assert response.status_code == 200
assert response.content == b"installer-bytes"
assert response.headers["content-type"] == "application/x-apple-diskimage"
assert response.headers["content-length"] == "15"
assert response.headers["content-disposition"] == (
'attachment; filename="OpenSwarm-arm64-affiliate_hash_123.dmg"'
)
assert response.headers["cache-control"] == "private, no-store"
assert json.loads(calls[0].content) == {"platform": "mac", "arch": "arm64", "ref": "alice"}
assert calls[1].url.host == "github.test"


def test_download_rejects_app_subdomains_without_upstream_call(monkeypatch):
calls = _mock_download_clients(monkeypatch)
client = TestClient(edge_main.app)
response = client.get(
"/download/mac/arm64",
headers={"host": "notes.openswarm.host", "fly-client-ip": "198.51.100.12"},
)

assert response.status_code == 404
assert calls == []


def test_download_maps_unknown_target_to_404(monkeypatch):
_mock_download_clients(monkeypatch, reject_unknown=True)
client = TestClient(edge_main.app)
response = client.get(
"/download/plan9/x64",
headers={"host": edge_main.EDGE_PUBLIC_HOST, "fly-client-ip": "198.51.100.13"},
)

assert response.status_code == 404
assert response.json() == {"error": "unknown download target"}


def test_download_rate_limits_per_client_ip(monkeypatch):
calls = _mock_download_clients(monkeypatch)
monkeypatch.setattr(edge_main, "_download_limiter", RateLimiter(limit=1, window_seconds=60))
client = TestClient(edge_main.app)
headers = {"host": edge_main.EDGE_PUBLIC_HOST, "fly-client-ip": "198.51.100.14"}

assert client.get("/download/mac/arm64", headers=headers).status_code == 200
assert client.get("/download/mac/arm64", headers=headers).status_code == 429
assert len(calls) == 2 # authorize + artifact only for the allowed request


def test_sandbox_rejects_unsafe_and_allows_safe():
try:
validate_code_safety("import os\nresult={}")
Expand Down
Loading