Expand Pingora gateway protocol examples

Author: fffonion

Cargo.toml

@@ -5,9 +5,9 @@ edition = "2024"
 license = "Apache-2.0"
 
 [dependencies]
-pd-host-function = { path = "../rustscript/pd-vm/pd-host-function" }
+pd-host-function = { path = "../rustscript/pd-host-function" }
 pingora = "0.8.1"
-vm = { package = "pd-vm", path = "../rustscript/pd-vm", default-features = false, features = ["runtime"] }
+vm = { package = "pd-vm", path = "../rustscript", default-features = false, features = ["runtime"] }
 
 [dev-dependencies]
 pretty_assertions = "1"

README.md

@@ -4,21 +4,30 @@ Standalone Pingora integration demo for `pd-vm` / RustScript.
 
 ## What it proves
 
-A Pingora gateway can keep the framework and proxy code compiled while moving request/response policy into RustScript:
+A Pingora gateway can keep the framework and proxy code compiled while moving request, transport, protocol, and response policy into RustScript:
 
-- live `pingora::http::RequestHeader` reads: `pingora::request::method`, `pingora::request::path`, `pingora::request::header`
-- live Pingora mutations: `pingora::request::insert_header`, `pingora::response::set_status`, `pingora::response::insert_header`
-- host functions are exported with `#[pd_host_function(name = "pingora::...")]`; the `.rss` names and namespaces match the bound host names exactly
+- live `pingora::http::RequestHeader` reads and request mutation
+- HTTP request/response/exchange host functions compatible with the `pd-edge` examples
+- proxy stream functions for native forwarding, bridged forwarding, and exchange-backed streams
+- TCP, UDP, TLS termination, WebSocket, and WebRTC-style host functions for protocol policy demos
+- host function names are bound in the same namespaces used by the `.rss` scripts: `http`, `proxy`, `tcp`, `udp`, `tls`, `websocket`, `webrtc`, `rate_limit`, and compatibility `pingora::*` names
 
 This does not fork or patch Pingora. It depends on the upstream `pingora` crate plus local `pd-vm` / `pd-host-function` paths only.
 
+## Scripts
+
+| script | focus |
+|---|---|
+| `scripts/gateway_policy.rss` | direct Pingora request/response policy |
+| `scripts/http_proxy.rss` | HTTP exchange, streaming body reads, proxy forwarding, rate limit decision |
+| `scripts/tls_termination.rss` | raw transport prelude, TLS handshake, downstream HTTP attach |
+| `scripts/websocket_proxy.rss` | WebSocket upstream configuration, message round trip, stream bridge |
+| `scripts/transport_matrix.rss` | TCP, TLS, UDP, and WebRTC-style host calls in one policy |
+
 ## Run
 
 ```bash
 cargo test --tests --jobs 4
 cargo run --example decision
+cargo run --example protocols
 ```
-
-## Script
-
-See `scripts/gateway_policy.rss`.

examples/protocols.rs

@@ -0,0 +1,48 @@
+use pingora::http::RequestHeader;
+use rustscript_pingora_gateway_policy::ScriptedGatewayPolicy;
+
+fn run(label: &str, script: &str, path: &[u8], headers: &[(&str, &str)]) {
+    let policy = ScriptedGatewayPolicy::from_source(script).expect("script should compile");
+    let mut request = RequestHeader::build("GET", path, None).expect("request should build");
+    for (name, value) in headers {
+        request
+            .insert_header((*name).to_string(), (*value).to_string())
+            .expect("header should insert");
+    }
+    let result = policy
+        .evaluate(&mut request)
+        .expect("script should evaluate");
+    println!(
+        "{label}: status={} body={} proxy_events={}",
+        result.response.status.as_u16(),
+        result.body,
+        result.proxy_events.len()
+    );
+}
+
+fn main() {
+    run(
+        "http_proxy",
+        include_str!("../scripts/http_proxy.rss"),
+        b"/proxy?mode=stream",
+        &[("x-stream", "1"), ("x-client-id", "paid")],
+    );
+    run(
+        "tls_termination",
+        include_str!("../scripts/tls_termination.rss"),
+        b"/terminate?x=1",
+        &[("x-downstream-scheme", "tcp")],
+    );
+    run(
+        "websocket_proxy",
+        include_str!("../scripts/websocket_proxy.rss"),
+        b"/ws",
+        &[("x-ws-message", "hello-ws")],
+    );
+    run(
+        "transport_matrix",
+        include_str!("../scripts/transport_matrix.rss"),
+        b"/transport",
+        &[],
+    );
+}

scripts/http_proxy.rss

@@ -0,0 +1,36 @@
+use http;
+use proxy;
+use rate_limit;
+
+let path: string = http::request::get_path();
+let client: string = http::request::get_header("x-client-id");
+
+if !rate_limit::allow(client, 100, 60) {
+    http::response::set_status(429);
+    http::response::set_header("x-rustscript-action", "rate-limited");
+    http::response::set_body("rate limited");
+} else {
+    let exchange: int = http::exchange::default_upstream();
+    http::exchange::set_target(exchange, "127.0.0.1", 18080);
+    http::exchange::set_path(exchange, path);
+    http::exchange::set_header(exchange, "x-forwarded-by", "rustscript-pingora");
+
+    if http::request::get_header("x-stream") != "" {
+        http::exchange::send(exchange);
+        http::response::set_status(http::exchange::get_status(exchange));
+        http::response::set_header("content-type", http::exchange::get_header(exchange, "content-type"));
+        let mut body = "";
+        while !http::exchange::body::eof(exchange) {
+            body = body + http::exchange::body::next_chunk(exchange, 8);
+        }
+        http::response::set_body(body);
+        http::response::set_header("x-rustscript-action", "streaming-proxy");
+    } else {
+        let downstream: int = proxy::stream::downstream();
+        let upstream: int = proxy::stream::exchange(exchange);
+        let status: string = proxy::forward(downstream, upstream, 16);
+        http::response::set_status(200);
+        http::response::set_header("x-proxy-status", status);
+        http::response::set_header("x-rustscript-action", "native-proxy");
+    }
+}

scripts/tls_termination.rss

@@ -0,0 +1,31 @@
+use http;
+use runtime;
+use tcp;
+use tls;
+
+let downstream: int = tcp::stream::downstream();
+let scheme: string = http::request::get_scheme();
+
+if scheme == "tcp" {
+    let prefix: string = tcp::stream::peek(downstream, 4);
+    if prefix == "GET " || prefix == "POST" || prefix == "HEAD" {
+        http::downstream::attach_transport();
+        http::response::set_header("x-termination", "plain-http");
+    } else {
+        let session: int = tls::session::from_socket(downstream);
+        tls::session::set_alpn(session, "h2,http/1.1");
+        tls::session::set_sni(session, "gateway.local");
+        if tls::session::handshake(session) {
+            http::downstream::attach_transport();
+            http::response::set_header("x-termination", "tls");
+            http::response::set_header("x-tls-alpn", tls::session::get_alpn(session));
+            http::response::set_header("x-tls-peer", tls::session::get_peer_name(session));
+        } else {
+            tcp::stream::close(downstream);
+            runtime::exit();
+        }
+    }
+}
+
+http::response::set_status(201);
+http::response::set_body(http::request::get_method() + "|" + http::request::get_path_with_query());

scripts/transport_matrix.rss

@@ -0,0 +1,34 @@
+use http;
+use tcp;
+use tls;
+use udp;
+use webrtc;
+
+let tcp_stream: int = tcp::stream::new();
+tcp::stream::set_target(tcp_stream, "127.0.0.1", 19000);
+tcp::stream::connect(tcp_stream);
+tcp::stream::write(tcp_stream, "hello-tcp");
+
+let session: int = tls::session::from_socket(tcp_stream);
+tls::session::set_verify(session, false);
+tls::session::set_alpn(session, "edge/1");
+tls::session::handshake(session);
+
+let socket: int = udp::socket::new();
+udp::socket::set_target(socket, "127.0.0.1", 19001);
+udp::socket::send_text(socket, "hello-udp");
+let udp_reply: string = udp::socket::recv_text(socket, 64);
+
+let rtc: int = webrtc::connection::new();
+webrtc::connection::set_ice_servers(rtc, "stun:localhost");
+webrtc::connection::set_data_channel_label(rtc, "policy");
+webrtc::connection::connect(rtc);
+webrtc::connection::send_text(rtc, "hello-rtc");
+
+http::response::set_status(200);
+http::response::set_header("x-tcp-phase", tcp::stream::get_phase(tcp_stream));
+http::response::set_header("x-tls-phase", tls::session::get_phase(session));
+http::response::set_header("x-tls-alpn", tls::session::get_alpn(session));
+http::response::set_header("x-udp-reply", udp_reply);
+http::response::set_header("x-webrtc-phase", webrtc::connection::get_phase(rtc));
+http::response::set_body("transport matrix ok");

scripts/websocket_proxy.rss

@@ -0,0 +1,30 @@
+use http;
+use proxy;
+use websocket;
+
+let downstream: int = websocket::connection::downstream();
+http::response::set_header("x-downstream-ws", websocket::connection::get_phase(downstream));
+
+let upstream: int = websocket::connection::default_upstream();
+websocket::connection::set_target(upstream, "127.0.0.1", 18091);
+websocket::connection::set_path(upstream, "/echo");
+websocket::connection::set_subprotocols(upstream, "chat,superchat");
+websocket::connection::connect(upstream);
+
+let mut message: string = http::request::get_header("x-ws-message");
+if message == "" {
+    message = "pingora-websocket";
+}
+websocket::connection::send_text(upstream, message);
+let reply: string = websocket::connection::read_text(upstream);
+
+let down_stream: int = proxy::stream::downstream();
+let up_stream: int = proxy::stream::from_websocket_binary(upstream);
+let proxy_status: string = proxy::bridge(down_stream, up_stream, 8);
+
+http::response::set_status(200);
+http::response::set_header("x-ws-protocol", websocket::connection::get_subprotocol(upstream));
+http::response::set_header("x-proxy-status", proxy_status);
+http::response::set_body(reply);
+websocket::connection::close(upstream, 1000, "complete");
+http::response::set_header("x-ws-phase", websocket::connection::get_phase(upstream));