Run code you didn't write — with zero access to your credentials
Run untrusted, user-supplied code in an isolated WebAssembly sandbox — no host filesystem, no direct network, and credentials that never enter the code you don't control.
# TRUSTED — holds the connection, runs on a trusted worker
@app.service
class UserDirectory(BaseService):
def __init__(self, connectors):
self._db = connectors.get("postgres") # never leaves this worker
async def fetch_user(self, user_id: int) -> dict:
row = await self._db.execute(
"SELECT id, name FROM users WHERE id = %s", (user_id,)
)
return dict(row.fetchone())
# UNTRUSTED — runs in the sandbox: no DB handle, no keys, no network
@app.step(sandboxed=True)
async def score_user(user_id: int, services=None):
user = await services["UserDirectory"].fetch_user(user_id) # only path out
return {"user_id": user_id, "score": len(user["name"]) * 10}DigitalFrontier Flow Scripts features
Credentials never enter untrusted code
Database connections, API keys and tokens live only inside trusted Services on separate workers. Sandboxed code holds no key, no connection string and no token, so there is nothing in scope to exfiltrate even if the code is hostile.
# trusted worker holds it — never leaves
self._db = connectors.get("postgres")
@app.step(sandboxed=True) # no handle, no keys, no network
async def score(user_id, services=None): ...A WebAssembly capability sandbox, not a wrapper
Untrusted steps execute inside a WebAssembly isolation boundary with no access to the host filesystem, no direct outbound network and no process spawning. The sandbox is the authoritative control, enforced at runtime rather than by static scanning alone.
@app.step(sandboxed=True) # WebAssembly isolation boundary
async def transform(values, services=None):
... # no host fs, no direct network, no process spawnOnly the methods you approve
Sandboxed code can call only the specific public Service methods you expose — each call is bridged to a trusted worker that runs it with the real credentials and returns only the result. Least privilege by construction: no key, token or connection string is ever in the sandbox's reach.
# only path out — a public Service method,
# bridged to a trusted worker with the real credentials
user = await services["UserDirectory"].fetch_user(user_id)Declarative egress allow-listing
Declare exactly which hostnames, wildcard subdomains, CIDR ranges and ports a Service may reach; every other outbound connection is blocked at the socket layer, deny-by-default. A dry-run mode logs would-be violations without blocking, so you can roll network policy out safely before enforcing it.
@app.service(egress=["api.stripe.com:443", "*.internal.example.com", "10.0.0.0/8"])
class PaymentsAPI(BaseService): ...
# roll out safely first:
@app.service(egress=["api.internal.example.com"], egress_mode="dry-run")Layered admission with an honest boundary
Before user code is admitted it is checked by a static analysis pass that rejects a defined set of dangerous imports and builtins, and functions are cryptographically signed and verified before any deserialization. These are defense-in-depth layers on top of the runtime sandbox — not a substitute for it.
Runs as source, never deserialized
Sandboxed steps are shipped and executed as source code, never as pickled callables, which closes the untrusted-deserialization path. Combined with strict per-request tenant isolation — a request's identity is locked immutably once set and every data access is checked against that tenant's own resources — one tenant's code cannot reach another tenant's data.
@app.step(sandboxed=True) # shipped & run as source, never pickled
async def transform(values, services=None): ...
result = await app.run_transform(values=[1, 2, 3]).wait_result()Get early access to Flow Scripts.
Tell us about your workload and we'll get you into the preview.