orca_proxy/
lib.rs

1//! Reverse proxy with HTTP routing for containers and Wasm trigger dispatch.
2//!
3//! Routes HTTP traffic by `Host` header to container backends (round-robin),
4//! and by path pattern to Wasm component invocations via a callback.
5//! Supports automatic TLS via ACME/Let's Encrypt (Caddy-style zero-config).
6
7pub mod acme;
8mod acme_proxy;
9mod backend_timeouts;
10mod body;
11mod error_page;
12mod forward;
13mod handler;
14pub mod rate_limit;
15mod routing;
16mod security_headers;
17pub mod sni;
18pub mod tls;
19mod websocket;
20
21pub use orca_core::config::{FallbackConfig, SecurityHeadersConfig};
22/// Install the proxy's security-header policy (call once at startup). See
23/// [`security_headers`].
24pub use security_headers::init as init_security_headers;
25
26use std::collections::HashMap;
27use std::sync::Arc;
28use std::sync::atomic::AtomicUsize;
29
30use hyper::Request;
31use hyper::body::Incoming;
32use hyper::server::conn::http1;
33use hyper::service::service_fn;
34use hyper_util::rt::{TokioExecutor, TokioIo};
35use hyper_util::server::conn::auto;
36use tokio::net::TcpListener;
37use tokio::sync::RwLock;
38use tracing::{debug, info, warn};
39
40use acme::AcmeManager;
41pub use acme_proxy::{run_proxy_with_acme, run_proxy_with_acme_and_fallback};
42use handler::{handle_acme_challenge, handle_request};
43use rate_limit::RateLimiter;
44
45/// Largest HTTP/2 header list the TLS listener accepts. Matches what the
46/// HTTP/1.1 path tolerates in practice, so large SSO cookie jars work on both.
47const H2_MAX_HEADER_LIST: u32 = 64 * 1024;
48
49/// A backend target for container routing.
50#[derive(Debug, Clone)]
51pub struct RouteTarget {
52    /// Address in the form `ip:port`.
53    pub address: String,
54    /// Owning service name.
55    pub service_name: String,
56    /// Optional path pattern (e.g., `"/api/*"`). When `None`, this target is a
57    /// catch-all for the domain. When `Some`, only requests whose path matches
58    /// the pattern are routed here. Longest-prefix match wins.
59    pub path_pattern: Option<String>,
60    /// Traffic weight (1-100, default 100). Used for weighted routing
61    /// during canary deployments. Higher weight = more traffic.
62    pub weight: u32,
63    /// Prefix to strip from the request path before forwarding upstream,
64    /// e.g. `"/admin"`. With `path_pattern = "/admin/*"` and
65    /// `strip_prefix = Some("/admin")`, a request for `/admin/users` is
66    /// forwarded as `/users` — same semantics as Caddy's `handle_path`.
67    pub strip_prefix: Option<String>,
68}
69
70/// A Wasm HTTP trigger: maps a path pattern to a Wasm runtime instance.
71#[derive(Debug, Clone)]
72pub struct WasmTrigger {
73    /// Path pattern (e.g., "/api/edge/*").
74    pub pattern: String,
75    /// Wasm runtime instance ID.
76    pub runtime_id: String,
77    /// Service name for logging.
78    pub service_name: String,
79}
80
81/// Callback invoked when a request matches a Wasm trigger.
82/// Receives (runtime_id, method, path, body) and returns the response body string.
83pub type WasmInvoker =
84    Arc<dyn Fn(String, String, String, String) -> WasmInvokeFuture + Send + Sync>;
85
86/// Future type returned by the Wasm invoker.
87pub type WasmInvokeFuture =
88    std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, String>> + Send>>;
89
90/// Shared Wasm trigger table type.
91pub type SharedWasmTriggers = Arc<RwLock<Vec<WasmTrigger>>>;
92
93/// Run the reverse proxy on the given port.
94pub async fn run_proxy(
95    route_table: Arc<RwLock<HashMap<String, Vec<RouteTarget>>>>,
96    wasm_triggers: SharedWasmTriggers,
97    wasm_invoker: Option<WasmInvoker>,
98    port: u16,
99    tls_acceptor: Option<tokio_rustls::TlsAcceptor>,
100    acme_manager: Option<AcmeManager>,
101) -> anyhow::Result<()> {
102    let addr = format!("0.0.0.0:{port}");
103    let listener = TcpListener::bind(&addr).await?;
104    let proto = if tls_acceptor.is_some() {
105        "HTTPS"
106    } else {
107        "HTTP"
108    };
109    info!("Reverse proxy listening on {addr} ({proto})");
110
111    serve_loop(
112        listener,
113        route_table,
114        wasm_triggers,
115        wasm_invoker,
116        tls_acceptor,
117        acme_manager,
118    )
119    .await
120}
121
122/// Run the proxy with optional fallback support.
123#[allow(clippy::too_many_arguments)]
124pub async fn run_proxy_with_fallback(
125    route_table: Arc<RwLock<HashMap<String, Vec<RouteTarget>>>>,
126    wasm_triggers: SharedWasmTriggers,
127    wasm_invoker: Option<WasmInvoker>,
128    port: u16,
129    tls_acceptor: Option<tokio_rustls::TlsAcceptor>,
130    acme_manager: Option<AcmeManager>,
131    fallback: Option<FallbackConfig>,
132) -> anyhow::Result<()> {
133    let addr = format!("0.0.0.0:{port}");
134    let listener = TcpListener::bind(&addr).await?;
135    let proto = if tls_acceptor.is_some() {
136        "HTTPS"
137    } else {
138        "HTTP"
139    };
140    info!("Reverse proxy listening on {addr} ({proto})");
141
142    serve_loop_with_fallback(
143        listener,
144        route_table,
145        wasm_triggers,
146        wasm_invoker,
147        tls_acceptor,
148        acme_manager,
149        fallback,
150    )
151    .await
152}
153
154/// Shared dynamic cert resolver for hot-provisioning.
155pub type SharedCertResolver = Arc<acme::DynCertResolver>;
156
157/// Core accept loop shared by HTTP and HTTPS listeners.
158async fn serve_loop(
159    listener: TcpListener,
160    route_table: Arc<RwLock<HashMap<String, Vec<RouteTarget>>>>,
161    wasm_triggers: SharedWasmTriggers,
162    wasm_invoker: Option<WasmInvoker>,
163    tls_acceptor: Option<tokio_rustls::TlsAcceptor>,
164    acme_manager: Option<AcmeManager>,
165) -> anyhow::Result<()> {
166    serve_loop_with_fallback(
167        listener,
168        route_table,
169        wasm_triggers,
170        wasm_invoker,
171        tls_acceptor,
172        acme_manager,
173        None,
174    )
175    .await
176}
177
178/// Serve loop variant with fallback support for SNI passthrough and HTTP forwarding.
179#[allow(clippy::too_many_arguments)]
180pub(crate) async fn serve_loop_with_fallback(
181    listener: TcpListener,
182    route_table: Arc<RwLock<HashMap<String, Vec<RouteTarget>>>>,
183    wasm_triggers: SharedWasmTriggers,
184    wasm_invoker: Option<WasmInvoker>,
185    tls_acceptor: Option<tokio_rustls::TlsAcceptor>,
186    acme_manager: Option<AcmeManager>,
187    fallback: Option<FallbackConfig>,
188) -> anyhow::Result<()> {
189    let counter = Arc::new(AtomicUsize::new(0));
190    // Upload- and response-aware timeouts (#187), not reqwest's read_timeout.
191    let client = Arc::new(backend_timeouts::client());
192    // A TLS endpoint exists if this listener terminates TLS itself, or if
193    // an ACME manager is present — the plain-HTTP listener of the ACME
194    // dual-listener setup carries one for HTTP-01 challenges, which is
195    // exactly the "HTTPS runs alongside on 443" signal. With neither, the
196    // HTTP→HTTPS redirect has nowhere to send clients and must not fire
197    // (#123: ACME unconfigured meant every routed host redirected into a
198    // closed port).
199    let https_enabled = tls_acceptor.is_some() || acme_manager.is_some();
200    let acme = acme_manager.map(Arc::new);
201    let is_tls = tls_acceptor.is_some();
202    let rate_limiter = RateLimiter::new();
203
204    let fallback = Arc::new(fallback);
205    loop {
206        let (stream, peer) = match listener.accept().await {
207            Ok(conn) => conn,
208            Err(e) => {
209                warn!("Proxy accept error: {e}");
210                continue;
211            }
212        };
213
214        let routes = route_table.clone();
215        let triggers = wasm_triggers.clone();
216        let invoker = wasm_invoker.clone();
217        let counter = counter.clone();
218        let client = client.clone();
219        let acme = acme.clone();
220        let tls = tls_acceptor.clone();
221        let rl = rate_limiter.clone();
222        let fb = fallback.clone();
223        let routes_for_sni = routes.clone();
224
225        let fb_for_service = fb.clone();
226        tokio::spawn(async move {
227            let service = service_fn(move |req: Request<Incoming>| {
228                let routes = routes.clone();
229                let triggers = triggers.clone();
230                let invoker = invoker.clone();
231                let counter = counter.clone();
232                let client = client.clone();
233                let acme = acme.clone();
234                let rl = rl.clone();
235                let fb = fb_for_service.clone();
236                async move {
237                    if let Some(resp) = handle_acme_challenge(&req, acme.as_deref()).await {
238                        return Ok(resp);
239                    }
240                    let mut resp = handle_request(
241                        req,
242                        &routes,
243                        &triggers,
244                        invoker.as_ref(),
245                        &counter,
246                        &client,
247                        is_tls,
248                        https_enabled,
249                        &rl,
250                        peer,
251                        fb.as_ref().as_ref(),
252                    )
253                    .await?;
254                    // Inject baseline security headers (add-if-absent; HSTS only
255                    // over TLS). No-op unless a policy was installed at startup.
256                    security_headers::apply(&mut resp, is_tls);
257                    Ok::<_, hyper::Error>(resp)
258                }
259            });
260            if let Some(acceptor) = tls {
261                let mut stream = stream;
262                // Peek SNI to decide between local TLS termination and pass-through
263                let sni = sni::peek_sni(&mut stream).await;
264                let should_passthrough = if let Some(ref host) = sni {
265                    let routes_lock = routes_for_sni.read().await;
266                    let known = routes_lock.contains_key(host);
267                    drop(routes_lock);
268                    !known && fb.as_ref().as_ref().and_then(|f| f.tls.as_ref()).is_some()
269                } else {
270                    false
271                };
272
273                if should_passthrough {
274                    let target = fb
275                        .as_ref()
276                        .as_ref()
277                        .and_then(|f| f.tls.clone())
278                        .expect("checked above");
279                    debug!(?sni, %target, "SNI passthrough");
280                    match tokio::net::TcpStream::connect(&target).await {
281                        Ok(mut backend) => {
282                            if let Err(e) =
283                                tokio::io::copy_bidirectional(&mut stream, &mut backend).await
284                            {
285                                debug!("Passthrough copy error from {peer}: {e}");
286                            }
287                        }
288                        Err(e) => warn!("Failed to connect to TLS fallback {target}: {e}"),
289                    }
290                    return;
291                }
292
293                match acceptor.accept(stream).await {
294                    Ok(tls_stream) => {
295                        let io = TokioIo::new(tls_stream);
296                        // h1 or h2, picked by the ALPN result. WebSocket
297                        // upgrades stay on h1: extended CONNECT (RFC 8441)
298                        // is not enabled, so browsers open a separate
299                        // HTTP/1.1 connection for them, as before.
300                        // hyper's h2 default caps the header list at 16 KiB and
301                        // answers 431 above it; SSO cookie jars (Keycloak plus
302                        // the app's own) exceed that where h1 never refused.
303                        let mut builder = auto::Builder::new(TokioExecutor::new());
304                        builder.http2().max_header_list_size(H2_MAX_HEADER_LIST);
305                        if let Err(e) = builder.serve_connection_with_upgrades(io, service).await {
306                            debug!("TLS proxy error from {peer}: {e}");
307                        }
308                    }
309                    Err(e) => debug!("TLS handshake failed from {peer}: {e}"),
310                }
311            } else {
312                // The plain listener stays HTTP/1-only: browsers never speak
313                // cleartext h2 (h2c), and this port only serves redirects,
314                // ACME challenges and plain-HTTP routes, so accepting h2c
315                // would add parser surface for no client.
316                let io = TokioIo::new(stream);
317                if let Err(e) = http1::Builder::new()
318                    .serve_connection(io, service)
319                    .with_upgrades()
320                    .await
321                {
322                    debug!("Proxy connection error from {peer}: {e}");
323                }
324            }
325        });
326    }
327}
orca_proxy/lib.rs

orca_proxy/
lib.rs
