diff --git a/openswarm-edge/.gitignore b/openswarm-edge/.gitignore new file mode 100644 index 00000000..77ac7549 --- /dev/null +++ b/openswarm-edge/.gitignore @@ -0,0 +1,3 @@ +.venv/ +__pycache__/ +*.pyc diff --git a/openswarm-edge/Dockerfile b/openswarm-edge/Dockerfile new file mode 100644 index 00000000..63dd9225 --- /dev/null +++ b/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}"] diff --git a/openswarm-edge/app/__init__.py b/openswarm-edge/app/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/openswarm-edge/app/bundles.py b/openswarm-edge/app/bundles.py new file mode 100644 index 00000000..652cf731 --- /dev/null +++ b/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 diff --git a/openswarm-edge/app/fallback.py b/openswarm-edge/app/fallback.py new file mode 100644 index 00000000..9c9a92c9 --- /dev/null +++ b/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"

{status_note}

" if status_note else "" + return ( + "" + "" + f"{title}" + f"
" + f"
🐝
" + f"

{title}

" + f"

{message}

" + f"{note}" + "

Build your own with OpenSwarm →

" + "
" + ) + + +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." diff --git a/openswarm-edge/app/main.py b/openswarm-edge/app/main.py new file mode 100644 index 00000000..9d4820ae --- /dev/null +++ b/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()) diff --git a/openswarm-edge/app/ratelimit.py b/openswarm-edge/app/ratelimit.py new file mode 100644 index 00000000..8a62e71e --- /dev/null +++ b/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 diff --git a/openswarm-edge/app/sandbox.py b/openswarm-edge/app/sandbox.py new file mode 100644 index 00000000..c7a0947e --- /dev/null +++ b/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__", "")) diff --git a/openswarm-edge/fly.toml b/openswarm-edge/fly.toml new file mode 100644 index 00000000..157a5bd9 --- /dev/null +++ b/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 diff --git a/openswarm-edge/requirements.txt b/openswarm-edge/requirements.txt new file mode 100644 index 00000000..e550411a --- /dev/null +++ b/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 diff --git a/openswarm-edge/tests/test_edge.py b/openswarm-edge/tests/test_edge.py new file mode 100644 index 00000000..0889e915 --- /dev/null +++ b/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"home", + "assets/app.js": b"console.log(1)", + "backend.py": b"result={}", + })) + assert resolve_file(b, "/")[0] == b"home" + assert resolve_file(b, "assets/app.js")[1] == "text/javascript" + assert resolve_file(b, "deep/spa/route")[0] == b"home" # SPA fallback + assert resolve_file(b, "backend.py")[0] == b"home" # never serve source + assert resolve_file(b, "../../etc/passwd")[0] == b"home" # 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()