Merge pull request #5985 from cloudflare/gv/set-egress-binding-pr

container: Implement setEgressHttp for containers

Author: gabivlj

build/deps/gen/deps.MODULE.bazel

@@ -27,10 +27,10 @@ bazel_dep(name = "brotli", version = "1.2.0")
 # capnp-cpp
 http.archive(
     name = "capnp-cpp",
-    sha256 = "58a883721d220a3d8d75531a7e7ede3fd87c3d6923caf645faff0c78f8807b23",
-    strip_prefix = "capnproto-capnproto-79b3170/c++",
+    sha256 = "2e8519d77eb453463b1f2b1e22f40959fe560c143d78e7a51c606ce3bca30c5b",
+    strip_prefix = "capnproto-capnproto-ac7d90a/c++",
     type = "tgz",
-    url = "https://github.com/capnproto/capnproto/tarball/79b317039adad92da1204929f4047f84dfd17350",
+    url = "https://github.com/capnproto/capnproto/tarball/ac7d90ae2e171a714be4348718fed6f26b0b85f2",
 )
 use_repo(http, "capnp-cpp")
 

build/deps/oci.MODULE.bazel

@@ -12,4 +12,12 @@ oci.pull(
         "linux/arm64/v8",
     ],
 )
-use_repo(oci, "node_25_slim", "node_25_slim_linux_amd64", "node_25_slim_linux_arm64_v8")
+oci.pull(
+    name = "proxy_everything",
+    image = "docker.io/cloudflare/proxy-everything:main",
+    platforms = [
+        "linux/amd64",
+        "linux/arm64",
+    ],
+)
+use_repo(oci, "node_25_slim", "node_25_slim_linux_amd64", "node_25_slim_linux_arm64_v8", "proxy_everything", "proxy_everything_linux_amd64", "proxy_everything_linux_arm64")

images/BUILD.bazel

@@ -2,6 +2,7 @@ load("@rules_multirun//:defs.bzl", "command", "multirun")
 
 IMAGES = {
     "container-client-test": "//images/container-client-test:load",
+    "proxy-everything": "//images/container-client-test:load-proxy-everything",
 }
 
 [

images/container-client-test/BUILD.bazel

@@ -40,3 +40,10 @@ oci_load(
     repo_tags = ["cloudflare/workerd/container-client-test:latest"],
     visibility = ["//visibility:public"],
 )
+
+oci_load(
+    name = "load-proxy-everything",
+    image = "@proxy_everything",
+    repo_tags = ["cloudflare/proxy-everything:main"],
+    visibility = ["//visibility:public"],
+)

images/container-client-test/app.js

@@ -1,11 +1,10 @@
 const { createServer } = require('http');
 
 const webSocketEnabled = process.env.WS_ENABLED === 'true';
+const wsProxyTarget = process.env.WS_PROXY_TARGET || null;
 
-// Create HTTP server
 const server = createServer(function (req, res) {
   if (req.url === '/ws') {
-    // WebSocket upgrade will be handled by the WebSocket server
     return;
   }
 
@@ -34,6 +33,25 @@ const server = createServer(function (req, res) {
       res.write(`Error reading /proc/1/cmdline: ${err.message}`);
       res.end();
     }
+
+    return;
+  }
+
+  if (req.url === '/intercept') {
+    const targetHost = req.headers['x-host'] || '11.0.0.1';
+    fetch(`http://${targetHost}`)
+      .then((result) => result.text())
+      .then((body) => {
+        res.writeHead(200);
+        res.write(body);
+        res.end();
+      })
+      .catch((err) => {
+        res.writeHead(500);
+        res.write(`${targetHost} ${err.message}`);
+        res.end();
+      });
+
     return;
   }
 
@@ -42,30 +60,30 @@ const server = createServer(function (req, res) {
   res.end();
 });
 
-// Check if WebSocket functionality is enabled
 if (webSocketEnabled) {
   const WebSocket = require('ws');
+  const wss = new WebSocket.Server({ server, path: '/ws' });
 
-  // Create WebSocket server
-  const wss = new WebSocket.Server({
-    server: server,
-    path: '/ws',
-  });
-
-  wss.on('connection', function connection(ws) {
-    console.log('WebSocket connection established');
+  wss.on('connection', function (clientWs) {
+    if (wsProxyTarget) {
+      const targetWs = new WebSocket(`ws://${wsProxyTarget}/ws`);
+      const ready = new Promise(function (resolve) {
+        targetWs.on('open', resolve);
+      });
 
-    ws.on('message', function message(data) {
-      console.log('Received:', data.toString());
-      // Echo the message back with prefix
-      ws.send('Echo: ' + data.toString());
-    });
+      targetWs.on('message', (data) => clientWs.send(data));
+      clientWs.on('message', async function (data) {
+        await ready;
+        targetWs.send(data);
+      });
 
-    ws.on('close', function close() {
-      console.log('WebSocket connection closed');
-    });
-
-    ws.on('error', console.error);
+      clientWs.on('close', targetWs.close);
+      targetWs.on('close', clientWs.close);
+    } else {
+      clientWs.on('message', function (data) {
+        clientWs.send('Echo: ' + data.toString());
+      });
+    }
   });
 }
 

src/workerd/api/container.c++

@@ -73,6 +73,39 @@ jsg::Promise<void> Container::setInactivityTimeout(jsg::Lock& js, int64_t durati
   return IoContext::current().awaitIo(js, req.sendIgnoringResult());
 }
 
+jsg::Promise<void> Container::interceptOutboundHttp(
+    jsg::Lock& js, kj::String addr, jsg::Ref<Fetcher> binding) {
+  auto& ioctx = IoContext::current();
+  auto channel = binding->getSubrequestChannel(ioctx);
+
+  // Get a channel token for RPC usage, the container runtime can use this
+  // token later to redeem a Fetcher.
+  auto token = channel->getToken(IoChannelFactory::ChannelTokenUsage::RPC);
+
+  auto req = rpcClient->setEgressHttpRequest();
+  req.setHostPort(addr);
+  req.setChannelToken(token);
+  return ioctx.awaitIo(js, req.sendIgnoringResult());
+}
+
+jsg::Promise<void> Container::interceptAllOutboundHttp(jsg::Lock& js, jsg::Ref<Fetcher> binding) {
+  auto& ioctx = IoContext::current();
+  auto channel = binding->getSubrequestChannel(ioctx);
+  auto token = channel->getToken(IoChannelFactory::ChannelTokenUsage::RPC);
+
+  // Register for all IPv4 and IPv6 addresses (on port 80)
+  auto reqV4 = rpcClient->setEgressHttpRequest();
+  reqV4.setHostPort("0.0.0.0/0"_kj);
+  reqV4.setChannelToken(token);
+
+  auto reqV6 = rpcClient->setEgressHttpRequest();
+  reqV6.setHostPort("::/0"_kj);
+  reqV6.setChannelToken(token);
+
+  return ioctx.awaitIo(js,
+      kj::joinPromisesFailFast(kj::arr(reqV4.sendIgnoringResult(), reqV6.sendIgnoringResult())));
+}
+
 jsg::Promise<void> Container::monitor(jsg::Lock& js) {
   JSG_REQUIRE(running, Error, "monitor() cannot be called on a container that is not running.");
 

src/workerd/api/container.h

@@ -62,6 +62,9 @@ class Container: public jsg::Object {
   void signal(jsg::Lock& js, int signo);
   jsg::Ref<Fetcher> getTcpPort(jsg::Lock& js, int port);
   jsg::Promise<void> setInactivityTimeout(jsg::Lock& js, int64_t durationMs);
+  jsg::Promise<void> interceptOutboundHttp(
+      jsg::Lock& js, kj::String addr, jsg::Ref<Fetcher> binding);
+  jsg::Promise<void> interceptAllOutboundHttp(jsg::Lock& js, jsg::Ref<Fetcher> binding);
 
   // TODO(containers): listenTcp()
 
@@ -73,6 +76,11 @@ class Container: public jsg::Object {
     JSG_METHOD(signal);
     JSG_METHOD(getTcpPort);
     JSG_METHOD(setInactivityTimeout);
+
+    if (flags.getWorkerdExperimental()) {
+      JSG_METHOD(interceptOutboundHttp);
+      JSG_METHOD(interceptAllOutboundHttp);
+    }
   }
 
   void visitForMemoryInfo(jsg::MemoryTracker& tracker) const {

src/workerd/io/container.capnp

@@ -114,4 +114,13 @@ interface Container @0x9aaceefc06523bca {
   # Note that if there is an open connection to the container, the runtime must not shutdown the container.
   # If there is no activity timeout duration configured and no container connection, it's up to the runtime
   # to decide when to signal the container to exit.
+
+  setEgressHttp @8 (hostPort :Text, channelToken :Data);
+  # Configures egress HTTP routing for the container. When the container attempts to connect to the
+  # specified host:port, the connection should be routed back to the Workers runtime using the channel token.
+  # The format of hostPort can be '<ip|cidr>[':'<port>]'. If port is omitted, it's assumed to only cover port 80.
+  # This method does not support HTTPs yet.
+
+
+  # TODO: setEgressTcp
 }

src/workerd/server/BUILD.bazel

@@ -204,19 +204,29 @@ wd_cc_library(
     }),
 )
 
+wd_cc_library(
+    name = "channel-token",
+    srcs = ["channel-token.c++"],
+    hdrs = ["channel-token.h"],
+    deps = [
+        ":channel-token_capnp",
+        "//src/workerd/io",
+        "//src/workerd/util:entropy",
+    ],
+)
+
 wd_cc_library(
     name = "server",
     srcs = [
-        "channel-token.c++",
         "server.c++",
     ],
     hdrs = [
-        "channel-token.h",
         "server.h",
     ],
     deps = [
         ":actor-id-impl",
         ":alarm-scheduler",
+        ":channel-token",
         ":channel-token_capnp",
         ":container-client",
         ":facet-tree-index",
@@ -268,7 +278,9 @@ wd_cc_library(
     hdrs = ["container-client.h"],
     visibility = ["//visibility:public"],
     deps = [
+        ":channel-token",
         ":docker-api_capnp",
+        "//src/workerd/io",
         "//src/workerd/io:container_capnp",
         "//src/workerd/jsg",
         "@capnp-cpp//src/capnp/compat:http-over-capnp",