[eric] edge: openswarm-edge service for *.openswarm.dev (static serve + sandbox compute + LLM proxy)

openswarm-edge/.gitignore

@@ -0,0 +1,3 @@
+.venv/
+__pycache__/
+*.pyc

openswarm-edge/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.13-slim
+
+WORKDIR /srv
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app ./app
+
+ENV PORT=8080
+EXPOSE 8080
+
+# One worker is fine: the work is IO-bound (Tigris + cloud proxy) and the sandbox
+# compute spawns its own subprocess. Scale out via Fly machines, not workers.
+CMD ["sh", "-c", "uvicorn app.main:app --host 0.0.0.0 --port ${PORT}"]

openswarm-edge/app/bundles.py

@@ -0,0 +1,129 @@
+"""Fetch + cache published app bundles from Tigris (read-only key). A bundle is a
+single apps/{slug}/bundle.tar.gz object; we unpack it once and cache the per-file
+bytes keyed by slug with a short TTL so a republish shows up without a restart.
+Every path lookup is guarded against traversal and never serves Python source."""
+from __future__ import annotations
+
+import asyncio
+import io
+import mimetypes
+import os
+import posixpath
+import tarfile
+import time
+from dataclasses import dataclass
+from typing import Optional
+
+import boto3
+from botocore.config import Config
+from botocore.exceptions import ClientError
+
+_ENDPOINT = os.environ.get("TIGRIS_ENDPOINT", "https://fly.storage.tigris.dev")
+_BUCKET = os.environ.get("TIGRIS_BUCKET", "openswarm-apps")
+_TTL_SECONDS = int(os.environ.get("EDGE_BUNDLE_TTL_SECONDS", "120"))
+_MAX_CACHED_BUNDLES = int(os.environ.get("EDGE_BUNDLE_CACHE_MAX", "200"))
+_MAX_UNPACKED_BYTES = 100 * 1024 * 1024  # guard against a decompression bomb
+
+# Browsers are picky about these; mimetypes' OS table can disagree across distros.
+_MIME_OVERRIDE = {
+    ".js": "text/javascript",
+    ".mjs": "text/javascript",
+    ".css": "text/css",
+    ".json": "application/json",
+    ".svg": "image/svg+xml",
+    ".wasm": "application/wasm",
+    ".map": "application/json",
+}
+
+_client = None
+
+
+def _s3():
+    global _client
+    if _client is None:
+        _client = boto3.client(
+            "s3",
+            endpoint_url=_ENDPOINT,
+            region_name=os.environ.get("TIGRIS_REGION", "auto"),
+            aws_access_key_id=os.environ.get("TIGRIS_ACCESS_KEY_ID", ""),
+            aws_secret_access_key=os.environ.get("TIGRIS_SECRET_ACCESS_KEY", ""),
+            config=Config(signature_version="s3v4"),
+        )
+    return _client
+
+
+@dataclass
+class Bundle:
+    files: dict[str, bytes]
+    backend_code: Optional[str]
+    fetched_at: float
+
+
+_cache: dict[str, Bundle] = {}
+
+
+def _bundle_key(slug: str) -> str:
+    return f"apps/{slug}/bundle.tar.gz"
+
+
+def unpack(tar_gz: bytes) -> Bundle:
+    files: dict[str, bytes] = {}
+    total = 0
+    with tarfile.open(fileobj=io.BytesIO(tar_gz), mode="r:gz") as tar:
+        for m in tar.getmembers():
+            if not m.isfile():
+                continue
+            name = posixpath.normpath(m.name).lstrip("/")
+            if name.startswith("..") or os.path.isabs(name):
+                continue
+            extracted = tar.extractfile(m)
+            if extracted is None:
+                continue
+            data = extracted.read()
+            total += len(data)
+            if total > _MAX_UNPACKED_BYTES:
+                raise ValueError("bundle exceeds the unpacked-size limit")
+            files[name] = data
+    backend = files.get("backend.py")
+    backend_code = backend.decode("utf-8", errors="replace") if backend is not None else None
+    return Bundle(files=files, backend_code=backend_code, fetched_at=time.time())
+
+
+async def get_bundle(slug: str) -> Optional[Bundle]:
+    cached = _cache.get(slug)
+    if cached and time.time() - cached.fetched_at < _TTL_SECONDS:
+        return cached
+    try:
+        obj = await asyncio.to_thread(lambda: _s3().get_object(Bucket=_BUCKET, Key=_bundle_key(slug)))
+        raw = await asyncio.to_thread(obj["Body"].read)
+    except ClientError as e:
+        code = str(e.response.get("Error", {}).get("Code", ""))
+        if code in ("NoSuchKey", "404", "NoSuchBucket", "AccessDenied"):
+            _cache.pop(slug, None)
+            return None
+        raise
+    bundle = unpack(raw)
+    if len(_cache) >= _MAX_CACHED_BUNDLES:
+        oldest = min(_cache, key=lambda k: _cache[k].fetched_at)
+        _cache.pop(oldest, None)
+    _cache[slug] = bundle
+    return bundle
+
+
+def resolve_file(bundle: Bundle, path: str) -> Optional[tuple[bytes, str]]:
+    """Map a request path to a bundle file, SPA-falling back to index.html. Refuses
+    traversal and Python source (served as index.html instead, never as code)."""
+    rel = posixpath.normpath(path.lstrip("/"))
+    if rel in ("", "."):
+        rel = "index.html"
+    if rel.startswith("..") or rel.endswith(".py"):
+        rel = "index.html"
+    data = bundle.files.get(rel)
+    if data is None:
+        data = bundle.files.get("index.html")
+        rel = "index.html"
+    if data is None:
+        return None
+    ext = posixpath.splitext(rel)[1].lower()
+    mime = _MIME_OVERRIDE.get(ext) or mimetypes.guess_type(rel)[0] or "application/octet-stream"
+    return data, mime

openswarm-edge/app/fallback.py

@@ -0,0 +1,46 @@
+"""Plain branded HTML for the states a visitor can hit that aren't the app itself:
+the slug isn't published (or was taken down), or the page is loaded on the apex
+instead of a {slug}.openswarm.dev subdomain. Kept inline + dependency-free so the
+edge can answer even when it can't reach storage."""
+from __future__ import annotations
+
+_STYLE = (
+    "font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;"
+    "background:#faf9f7;color:#2b2b2b;margin:0;min-height:100vh;display:flex;"
+    "align-items:center;justify-content:center;text-align:center;padding:24px;"
+)
+
+
+def _page(title: str, message: str, status_note: str = "") -> str:
+    note = f"<p style='color:#8a857c;font-size:13px;margin-top:24px'>{status_note}</p>" if status_note else ""
+    return (
+        "<!doctype html><html><head><meta charset='utf-8'>"
+        "<meta name='viewport' content='width=device-width,initial-scale=1'>"
+        f"<title>{title}</title></head>"
+        f"<body style=\"{_STYLE}\"><div style='max-width:420px'>"
+        f"<div style='font-size:40px;margin-bottom:12px'>🐝</div>"
+        f"<h1 style='font-size:20px;font-weight:650;margin:0 0 8px'>{title}</h1>"
+        f"<p style='color:#615d57;font-size:15px;line-height:1.5;margin:0'>{message}</p>"
+        f"{note}"
+        "<p style='margin-top:28px'><a href='https://openswarm.com' "
+        "style='color:#c2410c;text-decoration:none;font-weight:600'>Build your own with OpenSwarm &rarr;</a></p>"
+        "</div></body></html>"
+    )
+
+
+def not_found_page() -> str:
+    return _page(
+        "App not found",
+        "There's no published app at this address. It may have been unpublished or never existed.",
+    )
+
+
+def apex_page() -> str:
+    return _page(
+        "OpenSwarm Apps",
+        "Apps published from OpenSwarm live at their own subdomain here.",
+    )
+
+
+def report_footer_note() -> str:
+    return "See something off with an app? Report it in our Discord."

openswarm-edge/app/main.py

@@ -0,0 +1,156 @@
+"""openswarm-edge: the public face of {slug}.openswarm.dev.
+
+This service is intentionally the LEAST-privileged of the three: it holds only a
+read-only Tigris key + EDGE_SHARED_SECRET. It serves static app bundles, runs the
+sandboxed backend.py compute locally, and proxies runtime LLM calls to the cloud
+(which owns creator attribution, budgets, and pool credentials). The published
+page only ever talks to its own origin; the slug is derived here from the Host
+header and stamped on the internal call, so a page can never bill or impersonate
+another app."""
+from __future__ import annotations
+
+import os
+import re
+
+import httpx
+from fastapi import FastAPI, Request
+from fastapi.responses import HTMLResponse, JSONResponse, Response, StreamingResponse
+
+from .bundles import get_bundle, resolve_file
+from .fallback import apex_page, not_found_page
+from .ratelimit import RateLimiter
+from .sandbox import UnsafeCodeError, run_backend
+
+APPS_BASE_DOMAIN = os.environ.get("APPS_BASE_DOMAIN", "openswarm.dev")
+CLOUD_URL = os.environ.get("OPENSWARM_CLOUD_URL", "https://api.openswarm.com").rstrip("/")
+EDGE_SECRET = os.environ.get("EDGE_SHARED_SECRET", "")
+
+_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,31}$")
+_llm_limiter = RateLimiter(limit=30, window_seconds=60)
+_compute_limiter = RateLimiter(limit=60, window_seconds=60)
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+
+def slug_from_host(host: str) -> str | None:
+    """Extract the app slug from a {slug}.openswarm.dev Host header. Rejects the
+    apex, www, multi-label subdomains, and anything not slug-shaped."""
+    host = (host or "").split(":")[0].lower()
+    suffix = "." + APPS_BASE_DOMAIN
+    if not host.endswith(suffix):
+        return None
+    sub = host[: -len(suffix)]
+    if not sub or "." in sub or sub == "www":
+        return None
+    return sub if _SLUG_RE.match(sub) else None
+
+
+def client_ip(request: Request) -> str:
+    return request.headers.get("fly-client-ip") or (request.client.host if request.client else "unknown")
+
+
+def _security_headers() -> dict[str, str]:
+    # The hard isolation is the separate apex (no openswarm.com cookies are reachable
+    # here). CSP is defense-in-depth: block embedding + sniffing. We deliberately do
+    # NOT lock script/connect-src, arbitrary apps need to run their own JS and call
+    # their own APIs; the pre-publish scan is what screens for abusive content.
+    return {
+        "X-Content-Type-Options": "nosniff",
+        "Content-Security-Policy": "frame-ancestors 'none'",
+        "Referrer-Policy": "no-referrer-when-downgrade",
+        "Cache-Control": "public, max-age=60",
+    }
+
+
+@app.get("/__edge/health")
+async def health() -> JSONResponse:
+    return JSONResponse({"ok": True})
+
+
+@app.post("/__compute")
+async def edge_compute(request: Request) -> Response:
+    slug = slug_from_host(request.headers.get("host", ""))
+    if not slug:
+        return JSONResponse({"error": "unknown app"}, status_code=404)
+    if not _compute_limiter.allow(client_ip(request)):
+        return JSONResponse({"error": "Too many requests, slow down."}, status_code=429)
+    bundle = await get_bundle(slug)
+    if bundle is None:
+        return JSONResponse({"error": "app not found"}, status_code=404)
+    if not bundle.backend_code:
+        return JSONResponse({"error": "this app has no compute backend"}, status_code=404)
+    try:
+        payload = await request.json()
+    except Exception:
+        payload = {}
+    raw_input = payload.get("input_data", payload) if isinstance(payload, dict) else {}
+    input_data = raw_input if isinstance(raw_input, dict) else {}
+    try:
+        res = await run_backend(bundle.backend_code, input_data)
+    except UnsafeCodeError:
+        return JSONResponse({"error": "this app's backend can't run here"}, status_code=400)
+    except Exception:
+        return JSONResponse({"error": "compute failed"}, status_code=500)
+    return JSONResponse({"result": res.result, "stdout": res.stdout})
+
+
+@app.post("/__llm")
+async def edge_llm(request: Request) -> Response:
+    slug = slug_from_host(request.headers.get("host", ""))
+    if not slug:
+        return JSONResponse({"error": "unknown app"}, status_code=404)
+    if not _llm_limiter.allow(client_ip(request)):
+        return JSONResponse(
+            {"type": "error", "error": {"type": "rate_limited", "message": "Too many requests."}},
+            status_code=429,
+        )
+    body = await request.body()
+    headers = {
+        "x-edge-secret": EDGE_SECRET,
+        "x-app-slug": slug,
+        "content-type": "application/json",
+    }
+    for k in ("anthropic-version", "anthropic-beta"):
+        v = request.headers.get(k)
+        if v:
+            headers[k] = v
+
+    client = httpx.AsyncClient(timeout=httpx.Timeout(60.0, read=None))
+    upstream_req = client.build_request("POST", f"{CLOUD_URL}/api/apps/internal/llm", headers=headers, content=body)
+    try:
+        upstream = await client.send(upstream_req, stream=True)
+    except httpx.HTTPError:
+        await client.aclose()
+        return JSONResponse(
+            {"type": "error", "error": {"type": "upstream_unreachable", "message": "This app's AI is unavailable."}},
+            status_code=502,
+        )
+
+    async def relay():
+        try:
+            async for chunk in upstream.aiter_raw():
+                yield chunk
+        finally:
+            await upstream.aclose()
+            await client.aclose()
+
+    return StreamingResponse(
+        relay(),
+        status_code=upstream.status_code,
+        media_type=upstream.headers.get("content-type", "application/json"),
+    )
+
+
+@app.get("/{path:path}")
+async def serve_static(path: str, request: Request) -> Response:
+    slug = slug_from_host(request.headers.get("host", ""))
+    if not slug:
+        return HTMLResponse(apex_page(), status_code=404)
+    bundle = await get_bundle(slug)
+    if bundle is None:
+        return HTMLResponse(not_found_page(), status_code=404)
+    resolved = resolve_file(bundle, path)
+    if resolved is None:
+        return HTMLResponse(not_found_page(), status_code=404)
+    data, mime = resolved
+    return Response(content=data, media_type=mime, headers=_security_headers())

openswarm-edge/app/ratelimit.py

@@ -0,0 +1,40 @@
+"""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."""
+from __future__ import annotations
+
+import time
+
+# Hard ceiling on tracked keys so a flood of unique IPs can't grow the map without
+# bound; past it we drop the whole window (everyone gets a fresh allowance).
+_MAX_KEYS = 50_000
+
+
+class RateLimiter:
+    def __init__(self, limit: int, window_seconds: float):
+        self.limit = limit
+        self.window = window_seconds
+        self._hits: dict[str, list[float]] = {}
+
+    def allow(self, key: str) -> bool:
+        now = time.time()
+        if len(self._hits) > _MAX_KEYS:
+            self._hits.clear()
+        bucket = self._hits.get(key)
+        if bucket is None:
+            bucket = []
+            self._hits[key] = bucket
+        cutoff = now - self.window
+        # Drop timestamps that fell out of the window.
+        keep = 0
+        for t in bucket:
+            if t >= cutoff:
+                break
+            keep += 1
+        if keep:
+            del bucket[:keep]
+        if len(bucket) >= self.limit:
+            return False
+        bucket.append(now)
+        return True

openswarm-edge/app/sandbox.py

@@ -0,0 +1,123 @@
+"""Sandboxed Python runner for published apps' backend.py compute.
+
+VENDORED from backend/apps/outputs/executor.py (the desktop App Builder runtime).
+Keep the allow/deny lists + the subprocess hardening in sync with that file; this
+is the same data-shaping sandbox, just running in the edge instead of on the
+desktop. Pure compute only: no network, no disk, no subprocess, no secrets. Safe
+to run multi-tenant on one machine because nothing here can reach shared state."""
+from __future__ import annotations
+
+import ast
+import asyncio
+import json
+import os
+import sys
+import tempfile
+from dataclasses import dataclass
+
+TIMEOUT_SECONDS = 30
+
+_ALLOWED_MODULES = frozenset({
+    "json", "math", "re", "datetime", "collections", "itertools",
+    "functools", "statistics", "decimal", "fractions", "random",
+    "string", "textwrap", "unicodedata", "csv", "copy", "enum",
+    "dataclasses", "typing", "abc", "numbers", "uuid", "hashlib",
+    "base64", "binascii", "operator", "heapq", "bisect", "array",
+})
+
+_BLOCKED_BUILTINS = frozenset({
+    "exec", "eval", "compile", "__import__", "open", "input",
+    "breakpoint", "exit", "quit",
+})
+
+
+class UnsafeCodeError(Exception):
+    """AST validation rejected the backend code."""
+
+
+def validate_code_safety(code: str) -> None:
+    """Raise UnsafeCodeError on the first AST-visible risk. Published apps are
+    vetted at publish time, but we re-check here: the edge never trusts that the
+    bundle in storage matches what was scanned."""
+    try:
+        tree = ast.parse(code)
+    except SyntaxError as e:
+        raise UnsafeCodeError(f"Syntax error: {e}")
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                if alias.name.split(".")[0] not in _ALLOWED_MODULES:
+                    raise UnsafeCodeError(f"import '{alias.name}' is not allowed")
+        elif isinstance(node, ast.ImportFrom):
+            if node.module and node.module.split(".")[0] not in _ALLOWED_MODULES:
+                raise UnsafeCodeError(f"import from '{node.module}' is not allowed")
+        elif isinstance(node, ast.Call):
+            if isinstance(node.func, ast.Name) and node.func.id in _BLOCKED_BUILTINS:
+                raise UnsafeCodeError(f"builtin '{node.func.id}()' is not allowed")
+
+
+def _minimal_env() -> dict:
+    return {
+        "PYTHONDONTWRITEBYTECODE": "1",
+        "LANG": os.environ.get("LANG", "C.UTF-8"),
+        "LC_ALL": os.environ.get("LC_ALL", "C.UTF-8"),
+        "PYTHONUTF8": "1",
+        "PYTHONIOENCODING": "utf-8",
+    }
+
+
+@dataclass
+class ComputeResult:
+    result: dict
+    stdout: str
+
+
+async def run_backend(code: str, input_data: dict) -> ComputeResult:
+    """Validate + execute user backend code in a hardened subprocess. The code
+    reads `input_data` (a global dict) and assigns a global `result` dict."""
+    validate_code_safety(code)
+
+    preamble = (
+        "import json, sys, io, builtins\n"
+        "for _b in ('exec','eval','compile','open','input',\n"
+        "           'breakpoint','exit','quit'):\n"
+        "    try: delattr(builtins, _b)\n"
+        "    except AttributeError: pass\n"
+        "_orig_stdout = sys.stdout\n"
+        "_capture = io.StringIO()\n"
+        "sys.stdout = _capture\n"
+        "input_data = json.loads(sys.stdin.read())\n"
+        "result = {}\n"
+    )
+    postamble = (
+        "\nsys.stdout = _orig_stdout\n"
+        'json.dump({"__stdout__": _capture.getvalue(), "__result__": result}, sys.stdout)\n'
+    )
+    wrapper = preamble + code + postamble
+
+    with tempfile.TemporaryDirectory(prefix="osw-edge-exec-") as workdir:
+        proc = await asyncio.create_subprocess_exec(
+            sys.executable, "-c", wrapper,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            cwd=workdir,
+            env=_minimal_env(),
+        )
+        try:
+            stdout, stderr = await asyncio.wait_for(
+                proc.communicate(input=json.dumps(input_data).encode()),
+                timeout=TIMEOUT_SECONDS,
+            )
+        except asyncio.TimeoutError:
+            proc.kill()
+            await proc.wait()
+            raise RuntimeError(f"compute timed out after {TIMEOUT_SECONDS}s")
+
+    if proc.returncode != 0:
+        raise RuntimeError(f"compute failed: {stderr.decode(errors='replace').strip()[:500]}")
+    try:
+        parsed = json.loads(stdout.decode())
+    except json.JSONDecodeError:
+        raise RuntimeError("compute did not return valid JSON")
+    return ComputeResult(result=parsed.get("__result__", {}), stdout=parsed.get("__stdout__", ""))

openswarm-edge/fly.toml

@@ -0,0 +1,38 @@
+# openswarm-edge: serves *.openswarm.dev. Public-facing, least-privileged (only a
+# READ-ONLY Tigris key + EDGE_SHARED_SECRET, set via `fly secrets`, never here).
+# Wildcard cert: `fly certs create "*.openswarm.dev" -a openswarm-edge`.
+app = 'openswarm-edge'
+primary_region = 'iad'
+kill_signal = 'SIGINT'
+kill_timeout = '30s'
+
+[build]
+  dockerfile = 'Dockerfile'
+
+[env]
+  PORT = '8080'
+  APPS_BASE_DOMAIN = 'openswarm.dev'
+  OPENSWARM_CLOUD_URL = 'https://api.openswarm.com'
+  TIGRIS_ENDPOINT = 'https://fly.storage.tigris.dev'
+  TIGRIS_BUCKET = 'openswarm-apps'
+
+[http_service]
+  internal_port = 8080
+  force_https = true
+  auto_stop_machines = 'off'
+  min_machines_running = 1
+  [http_service.concurrency]
+    type = 'requests'
+    hard_limit = 250
+    soft_limit = 200
+  [[http_service.checks]]
+    interval = '30s'
+    timeout = '5s'
+    grace_period = '10s'
+    method = 'get'
+    path = '/__edge/health'
+
+[[vm]]
+  cpu_kind = 'shared'
+  cpus = 1
+  memory_mb = 512

openswarm-edge/requirements.txt

@@ -0,0 +1,4 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+boto3==1.35.90
+httpx==0.28.1

openswarm-edge/tests/test_edge.py

@@ -0,0 +1,92 @@
+"""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.
+
+Run with:  .venv/bin/python -m pytest tests/test_edge.py
+"""
+import asyncio
+import io
+import os
+import sys
+import tarfile
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from app.main import slug_from_host
+from app.bundles import unpack, resolve_file
+from app.ratelimit import RateLimiter
+from app.sandbox import validate_code_safety, run_backend, UnsafeCodeError
+
+
+def test_slug_from_host():
+    assert slug_from_host("notes.openswarm.dev") == "notes"
+    assert slug_from_host("notes.openswarm.dev:443") == "notes"
+    assert slug_from_host("UPPER.openswarm.dev") == "upper"
+    assert slug_from_host("openswarm.dev") is None        # apex
+    assert slug_from_host("www.openswarm.dev") is None     # www
+    assert slug_from_host("a.b.openswarm.dev") is None     # multi-label
+    assert slug_from_host("notes.evil.com") is None        # wrong domain
+    assert slug_from_host("bad_slug.openswarm.dev") is None  # underscore
+
+
+def _mk_tar(files: dict[str, bytes]) -> bytes:
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as t:
+        for name, data in files.items():
+            info = tarfile.TarInfo(name=name)
+            info.size = len(data)
+            t.addfile(info, io.BytesIO(data))
+    return buf.getvalue()
+
+
+def test_resolve_file_paths():
+    b = unpack(_mk_tar({
+        "index.html": b"<html>home</html>",
+        "assets/app.js": b"console.log(1)",
+        "backend.py": b"result={}",
+    }))
+    assert resolve_file(b, "/")[0] == b"<html>home</html>"
+    assert resolve_file(b, "assets/app.js")[1] == "text/javascript"
+    assert resolve_file(b, "deep/spa/route")[0] == b"<html>home</html>"   # SPA fallback
+    assert resolve_file(b, "backend.py")[0] == b"<html>home</html>"       # never serve source
+    assert resolve_file(b, "../../etc/passwd")[0] == b"<html>home</html>"  # traversal blocked
+
+
+def test_backend_code_available_for_compute_not_static():
+    b = unpack(_mk_tar({"index.html": b"x", "backend.py": b"import math\nresult={}"}))
+    assert b.backend_code == "import math\nresult={}"
+    data, _ = resolve_file(b, "backend.py")
+    assert data == b"x"
+
+
+def test_rate_limiter():
+    rl = RateLimiter(limit=3, window_seconds=100)
+    assert all(rl.allow("ip1") for _ in range(3))
+    assert rl.allow("ip1") is False  # 4th over the limit
+    assert rl.allow("ip2") is True   # a different key is independent
+
+
+def test_sandbox_rejects_unsafe_and_allows_safe():
+    try:
+        validate_code_safety("import os\nresult={}")
+        assert False, "expected UnsafeCodeError"
+    except UnsafeCodeError:
+        pass
+    validate_code_safety("import math\nresult={'x': math.pi}")  # no raise
+
+
+def test_sandbox_runs_safe_code():
+    res = asyncio.run(run_backend("result = {'sum': sum(input_data['nums'])}", {"nums": [1, 2, 3]}))
+    assert res.result == {"sum": 6}
+
+
+def _run_all():
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+    for fn in fns:
+        fn()
+        print(f"ok  {fn.__name__}")
+    print(f"\n{len(fns)} passed")
+
+
+if __name__ == "__main__":
+    _run_all()